package Ġnet . h ane k awa . p api ka . common . slack Ċ Ċ import Ġcom . squareup . mos hi . Json DataException Ċ import Ġcom . squareup . mos hi . M oshi Ċ import Ġcom . t img roup . stats d . Stats D Client Ċ import Ġnet . h ane k awa . p api ka . common . getLogger Ċ import Ġokhttp 3 . Request Ċ import Ġokhttp 3 . Response Ċ import Ġokhttp 3 . WebSocket Ċ import Ġokhttp 3 . WebSocket Listener ĊĊ Ċ interface ĠR tm EventHandler Ġ{ ĊĠĠĠ Ġfun Ġon Event ( web Socket : ĠWebSocket , Ġevent : ĠR tm Event ) ĊĠĠĠ Ġfun Ġon Failure ( web Socket : ĠWebSocket , Ġt : ĠThrowable , Ġresponse : ĠResponse ?) ĊĠĠĠ Ġfun Ġon Closing ( web Socket : ĠWebSocket ?, Ġcode : ĠInt , Ġreason : ĠString ?) Ċ } Ċ Ċ data Ġclass ĠPing Event ( ĊĠĠĠĠĠĠĠ Ġval Ġtype : ĠString Ġ= Ġ" ping ", ĊĠĠĠĠĠĠĠ Ġval Ġid : ĠString , ĊĠĠĠĠĠĠĠ Ġval Ġtime : ĠString Ċ ) ĊĊ Ċ class ĠJson Parsing Listener ( private Ġval Ġstats d : ĠStats D Client , Ġprivate Ġval Ġmessage Handler : ĠR tm EventHandler ) Ġ: ĠWebSocket Listener () Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠLOG Ġ= Ġget Logger ( this :: class . java ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġm oshi Ġ= ĠM oshi ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. Builder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġprivate Ġval Ġmap Adapter Ġ= Ġm oshi . adapter ( Map :: class . java ) ĊĠĠĠ Ġprivate Ġvar Ġhealth Checker : ĠWebSocket Health Checking Thread ? Ġ= Ġnull ĊĊĠĠĠ Ġoverride Ġfun Ġon Open ( web Socket : ĠWebSocket , Ġresponse : ĠResponse ) Ġ{ ĊĠĠĠĠĠĠĠ ĠLOG . debug (" Opening Ġwebsocket Ġgot Ġresponse : Ġ{}", Ġresponse ) ĊĠĠĠĠĠĠĠ Ġstats d . increment (" websocket . num _ open ") ĊĊĠĠĠĠĠĠĠ Ġhealth Checker Ġ= ĠWebSocket Health Checking Thread ( stats d , Ġweb Socket ) ĊĠĠĠĠĠĠĠ Ġhealth Checker !!. start () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Message ( web Socket : ĠWebSocket , Ġtext : ĠString ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġstats d . increment (" websocket . num _ message ") ĊĊĠĠĠĠĠĠĠ Ġif Ġ( text Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSlack R tm Session . LOG . warn (" Got Ġnull Ġmessage : Ġ{}", Ġtext ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġparsed Message Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmap Adapter . fromJson ( text ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠJson DataException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . error (" Unable Ġto Ġparse : Ġ{}", Ġtext ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} Ġ?: Ġreturn ĊĊĠĠĠĠĠĠĠ ĠLOG . debug (" Got Ġevent : Ġ{}", Ġtext ) ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġval Ġcast ed Event Ġ= Ġparsed Message Ġas ? ĠMap < String , ĠAny > Ġ?: Ġreturn ĊĊĠĠĠĠĠĠĠ Ġif Ġ(! cast ed Event . containsKey (" type ")) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . warn (" Received Ġevent Ġwithout Ġtype : Ġ{}", Ġtext ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġevent Ġ= ĠR tm Event ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtype Ġ= Ġcast ed Event [" type "] Ġas ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpayload Ġ= Ġcast ed Event ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġif Ġ( event . type Ġ== Ġ" p ong ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhealth Checker !!. handle P ong ( event ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNever Ġexpose Ġping / p ongs Ġto Ġthe Ġactual Ġmessage Ġhandler ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Handler . on Event ( web Socket , Ġevent ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Closing ( web Socket : ĠWebSocket , Ġcode : ĠInt , Ġreason : ĠString ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġstats d . increment (" websocket . num _ closing ", Ġ" code :$ code ") ĊĠĠĠĠĠĠĠ ĠLOG . info (" Closing Ġwith Ġcode Ġ{} : Ġ{}", Ġcode , Ġreason ) ĊĠĠĠĠĠĠĠ Ġhealth Checker !!. should Run Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ Ġmessage Handler . on Closing ( web Socket , Ġcode , Ġreason ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġon Closing Ġmeans Ġthe Ġremote Ġpeer Ġwants Ġto Ġclose . ĠWe Ġhave Ġto Ġclose Ġour Ġside Ġtoo ĊĠĠĠĠĠĠĠ Ġweb Socket . close ( 1 0 0 0 , Ġnull ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Failure ( web Socket : ĠWebSocket , Ġt : ĠThrowable , Ġresponse : ĠResponse ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġstats d . increment (" websocket . num _ failure ") ĊĠĠĠĠĠĠĠ ĠLOG . error (" Unexpected Ġfailure : Ġ{}", Ġresponse , Ġt ) ĊĠĠĠĠĠĠĠ Ġhealth Checker !!. should Run Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ Ġmessage Handler . on Failure ( web Socket , Ġt , Ġresponse ) ĊĠĠĠ Ġ} Ċ } ĊĊ Ċ class ĠWebSocket Health Checking Thread ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġstats d : ĠStats D Client , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġweb Socket : ĠWebSocket , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġping Interval Ms : ĠLong Ġ= Ġ 5 _ 0 0 0 , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġun answer ed P ings Threshold : ĠInt Ġ= Ġ 5 Ċ ) Ġ: ĠThread () Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠLOG Ġ= Ġget Logger ( this :: class . java ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġm oshi Ġ= ĠM oshi ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. Builder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġprivate Ġval Ġping Event Adapter Ġ= Ġm oshi . adapter ( Ping Event :: class . java ) ĊĊĠĠĠ Ġvar Ġshould Run Ġ= Ġtrue ĊĠĠĠ Ġprivate Ġvar Ġmost Recent P ong Received At : ĠLong ? Ġ= Ġnull ĊĊĠĠĠ Ġoverride Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġun answer ed P ings Counter Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠ Ġwhile Ġ( should Run ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmost Recent Ping Sent At Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġping Event Ġ= ĠPing Event ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġid Ġ= Ġmost Recent Ping Sent At . toString (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtime Ġ= Ġmost Recent Ping Sent At . toString () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjson Ġ= Ġping Event Adapter . toJson ( ping Event ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . debug (" Sending Ġping : Ġ{}", Ġjson ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġweb Socket . send ( json ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠThread . sleep ( ping Interval Ms ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAs sess Ġhealth ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmost Recent P ong Ġ= Ġsynchronized ( this ) Ġ{ Ġmost Recent P ong Received At Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmost Recent P ong Ġ== Ġnull Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . warn (" No Ġp ong Ġreceived ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġhave Ġnot Ġreceived Ġa Ġp ong Ġat Ġall ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġun answer ed P ings Counter Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmost Recent P ong Ġ< Ġmost Recent Ping Sent At Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . warn (" P ong Ġwas Ġstale Ġ( ping ={}, Ġp ong ={}, Ġdiff ={ })", Ġmost Recent Ping Sent At , Ġmost Recent P ong , Ġmost Recent P ong Ġ- Ġmost Recent Ping Sent At ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġhave Ġnot Ġreceived Ġa Ġp ong Ġsince Ġour Ġlatest Ġping ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġun answer ed P ings Counter Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġun answer ed P ings Counter Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġstats d . gauge (" websocket . health check . num _ un answer ed _ p ings ", Ġun answer ed P ings Counter . to Long ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġun answer ed P ings Counter Ġ>= Ġun answer ed P ings Threshold Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . info (" Reached Ġmaximum Ġthreshold Ġof Ġun answer ed Ġp ings Ġ( {} ), Ġdeclaring Ġun healthy ", Ġun answer ed P ings Threshold ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Un healthy () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġun answer ed P ings Counter Ġ> Ġ 0 Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . info (" Un answer ed Ġp ings : Ġ{} Ġ/ Ġ{}", Ġun answer ed P ings Counter , Ġun answer ed P ings Threshold ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠLOG . info (" Health checker Ġshould Ġno Ġlonger Ġrun , Ġstopping ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġhandle P ong ( event : ĠR tm Event ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġping Time Ms Ġ= Ġevent . payload [" time "]. toString (). to Long OrNull () ĊĊĠĠĠĠĠĠĠ Ġif Ġ( ping Time Ms Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . warn (" Got Ġmalformed Ġp ong : Ġ{}", Ġevent . payload ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠOnly Ġupdate Ġwhen Ġthe Ġp ong Ġlooks Ġcorrect ĊĠĠĠĠĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmost Recent P ong Received At Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval ĠcurrentTime Ms Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠ Ġval Ġtime Since Ping Ms Ġ= ĠcurrentTime Ms Ġ- Ġping Time Ms ĊĠĠĠĠĠĠĠ Ġstats d . record Execution Time (" websocket . health check . ping _ p ong _ time _ diff ", Ġtime Since Ping Ms ) ĊĠĠĠĠĠĠĠ ĠLOG . debug (" Received Ġp ong Ġ{} Ġms Ġafter Ġping Ġ( ping ={}, Ġp ong ={ })", Ġtime Since Ping Ms , Ġping Time Ms , Ġmost Recent P ong Received At ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġon Un healthy () Ġ{ ĊĠĠĠĠĠĠĠ Ġshould Run Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ ĠLOG . info (" WebSocket Ġhas Ġbeen Ġdeclared Ġun healthy , Ġclosing ") ĊĠĠĠĠĠĠĠ Ġweb Socket . close ( 1 0 0 0 , Ġnull ) ĊĠĠĠ Ġ} Ċ } ĊĊ Ċ class ĠSlack R tm Session ( private Ġval Ġstats d : ĠStats D Client , Ġprivate Ġval Ġslack Client : ĠSlack Client , Ġprivate Ġval Ġevent Handler : ĠR tm EventHandler ) Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠLOG Ġ= Ġget Logger ( this :: class . java ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġr tm Start Response Ġ= Ġslack Client . getR tm Start Response () ĊĠĠĠĠĠĠĠ ĠLOG . info (" RT M ĠStart Ġresponse : Ġ{}", Ġr tm Start Response . self ) ĊĊĠĠĠĠĠĠĠ Ġslack Client . httpClient . new WebSocket ( Request . Builder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. url ( rtm Start Response . url ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build (), ĠJson Parsing Listener ( stats d , Ġevent Handler )) ĊĊĠĠĠĠĠĠĠ Ġ// ĠTell Ġthe Ġexecutor Ġto Ġshutdown Ġ( otherwise , Ġwe 're Ġwaiting Ġfor Ġit Ġto Ġtimeout ) ĊĠĠĠĠĠĠĠ Ġslack Client . httpClient . dispatcher (). executor Service (). shutdown () ĊĠĠĠ Ġ} Ċ Ċ }