/* Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 1 9 ĠDaniel ĠRe es Ġ< daniel . re es 1 8 @ gmail . com > Ċ Ġ* Ċ Ġ* ĠPermission Ġis Ġhereby Ġgranted , Ġfree Ġof Ġcharge , Ġto Ġany Ġperson Ġobtaining Ġa Ġcopy Ċ Ġ* Ġof Ġthis Ġsoftware Ġand Ġassociated Ġdocumentation Ġfiles Ġ( the Ġ" Software "), Ġto Ġdeal Ċ Ġ* Ġin Ġthe ĠSoftware Ġwithout Ġrestriction , Ġincluding Ġwithout Ġlimitation Ġthe Ġrights Ċ Ġ* Ġto Ġuse , Ġcopy , Ġmodify , Ġmerge , Ġpublish , Ġdistribute , Ġsublicense , Ġand / or Ġsell Ċ Ġ* Ġcopies Ġof Ġthe ĠSoftware , Ġand Ġto Ġpermit Ġpersons Ġto Ġwhom Ġthe ĠSoftware Ġis Ċ Ġ* Ġfurnished Ġto Ġdo Ġso , Ġsubject Ġto Ġthe Ġfollowing Ġconditions : Ċ Ġ* Ċ Ġ* ĠThe Ġabove Ġcopyright Ġnotice Ġand Ġthis Ġpermission Ġnotice Ġshall Ġbe Ġincluded Ġin Ċ Ġ* Ġall Ġcopies Ġor Ġsubstantial Ġportions Ġof Ġthe ĠSoftware . Ċ Ġ* Ċ Ġ* ĠTHE ĠSOFTWARE ĠIS ĠPROVIDED Ġ" AS ĠIS ", ĠWITHOUT ĠWARRANTY ĠOF ĠANY ĠKIND , ĠEXPRESS ĠOR Ċ Ġ* ĠIMPLIED , ĠINCLUDING ĠBUT ĠNOT ĠLIMITED ĠTO ĠTHE ĠWARRANTIES ĠOF ĠMERCHANTABILITY , Ċ Ġ* ĠFITNESS ĠFOR ĠA ĠPARTICULAR ĠPURPOSE ĠAND ĠNONINFRINGEMENT . ĠIN ĠNO ĠEVENT ĠSHALL ĠTHE Ċ Ġ* ĠAUTHORS ĠOR ĠCOPYRIGHT ĠHOLDERS ĠBE ĠLIABLE ĠFOR ĠANY ĠCLAIM , ĠDAMAGES ĠOR ĠOTHER Ċ Ġ* ĠLIABILITY , ĠWHETHER ĠIN ĠAN ĠACTION ĠOF ĠCONTRACT , ĠTORT ĠOR ĠOTHERWISE , ĠARISING ĠFROM , Ċ Ġ* ĠOUT ĠOF ĠOR ĠIN ĠCONNECTION ĠWITH ĠTHE ĠSOFTWARE ĠOR ĠTHE ĠUSE ĠOR ĠOTHER ĠDEALINGS ĠIN Ċ Ġ* ĠTHE ĠSOFTWARE . Ċ Ġ*/ Ċ Ċ package Ġorg . phoenix framework Ċ Ċ import Ġjava . util . concurrent . Concurrent Linked Queue Ċ Ċ /** Ċ Ġ* ĠRepresents Ġa Ġbinding Ġto Ġa ĠChannel Ġevent Ċ Ġ*/ Ċ data Ġclass ĠBinding ( ĊĠ Ġval Ġevent : ĠString , ĊĠ Ġval Ġref : ĠInt , ĊĠ Ġval Ġcallback : Ġ( Message ) Ġ-> ĠUnit Ċ ) Ċ Ċ /** Ċ Ġ* ĠRepresents Ġa ĠChannel Ġbound Ġto Ġa Ġgiven Ġtopic Ċ Ġ*/ Ċ class ĠChannel ( ĊĠ Ġval Ġtopic : ĠString , ĊĠ Ġparams : ĠPayload , ĊĠ Ġinternal Ġval Ġsocket : ĠSocket Ċ ) Ġ{ ĊĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ// ĠChannel ĠNested ĠEnums ĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ/** ĠStates Ġof Ġa ĠChannel Ġ*/ ĊĠ Ġenum Ġclass ĠState () Ġ{ ĊĠĠĠ ĠCLO SED , ĊĠĠĠ ĠER RO RED , ĊĠĠĠ ĠJOIN ED , ĊĠĠĠ ĠJOIN ING , ĊĠĠĠ ĠLE AV ING ĊĠ Ġ} ĊĊĠ Ġ/** ĠChannel Ġspecific Ġevents Ġ*/ ĊĠ Ġenum Ġclass ĠEvent ( val Ġvalue : ĠString ) Ġ{ ĊĠĠĠ ĠHE ART BE AT (" heartbeat "), ĊĠĠĠ ĠJOIN (" ph x _ join "), ĊĠĠĠ ĠLE AVE (" ph x _ leave "), ĊĠĠĠ ĠRE PLY (" ph x _ reply "), ĊĠĠĠ ĠERROR (" ph x _ error "), ĊĠĠĠ ĠCLOSE (" ph x _ close "); ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠ Ġ/** ĠTrue Ġif Ġthe Ġevent Ġis Ġone Ġof ĠP ho enix 's Ġchannel Ġlifecycle Ġevents Ġ*/ ĊĠĠĠĠĠ Ġfun Ġis Lifecycle Event ( event : ĠString ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( event ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ ĠJOIN . value , ĊĠĠĠĠĠĠĠĠĠ ĠLE AVE . value , ĊĠĠĠĠĠĠĠĠĠ ĠRE PLY . value , ĊĠĠĠĠĠĠĠĠĠ ĠERROR . value , ĊĠĠĠĠĠĠĠĠĠ ĠCLOSE . value Ġ-> Ġtrue ĊĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ// ĠChannel ĠAttributes ĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ/** ĠCurrent Ġstate Ġof Ġthe ĠChannel Ġ*/ ĊĠ Ġinternal Ġvar Ġstate : ĠState ĊĊĠ Ġ/** ĠCollection Ġof Ġevent Ġbindings . Ġ*/ ĊĠ Ġinternal Ġval Ġbindings : ĠConcurrent Linked Queue < Binding > ĊĊĠ Ġ/** ĠTrack s Ġevent Ġbinding Ġref Ġcounters Ġ*/ ĊĠ Ġinternal Ġvar Ġbinding Ref : ĠInt ĊĊĠ Ġ/** ĠTimeout Ġwhen Ġattempting Ġto Ġjoin Ġa ĠChannel Ġ*/ ĊĠ Ġinternal Ġvar Ġtimeout : ĠLong ĊĊĠ Ġ/** ĠParams Ġpassed Ġin Ġthrough Ġcon structions Ġand Ġprovided Ġto Ġthe ĠJoin Push Ġ*/ ĊĠ Ġvar Ġparams : ĠPayload Ġ= Ġparams ĊĠĠĠ Ġset ( value ) Ġ{ ĊĠĠĠĠĠ Ġjoin Push . payload Ġ= Ġvalue ĊĠĠĠĠĠ Ġfield Ġ= Ġvalue ĊĠĠĠ Ġ} ĊĊĠ Ġ/** ĠSet Ġto Ġtrue Ġonce Ġthe Ġchannel Ġhas Ġattempted Ġto Ġjoin Ġ*/ ĊĠ Ġinternal Ġvar Ġjoined Once : ĠBoolean ĊĊĠ Ġ/** ĠPush Ġto Ġsend Ġthen Ġattempting Ġto Ġjoin Ġ*/ ĊĠ Ġinternal Ġvar Ġjoin Push : ĠPush ĊĊĠ Ġ/** ĠBuffer Ġof ĠP ushes Ġthat Ġwill Ġbe Ġsent Ġonce Ġthe ĠChannel 's Ġsocket Ġconnects Ġ*/ ĊĠ Ġinternal Ġvar Ġpush Buffer : ĠMutable List < Push > ĊĊĠ Ġ/** ĠTimer Ġto Ġattempt Ġre join s Ġ*/ ĊĠ Ġinternal Ġvar Ġre join Timer : ĠTimeout Timer ĊĊĠ Ġ/** ĠRe fs Ġif Ġstate Change Ġhooks Ġ*/ ĊĠ Ġinternal Ġvar Ġstate Change Refs : ĠMutable List < String > ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠOptional Ġon Message Ġhook Ġthat Ġcan Ġbe Ġprovided . ĠRe ceiv es Ġall Ġevent Ġmessages Ġfor Ġspecialized ĊĠĠ Ġ* Ġhandling Ġbefore Ġdispatch ing Ġto Ġthe ĠChannel Ġevent Ġcallbacks . ĊĠĠ Ġ*/ ĊĠ Ġinternal Ġvar Ġon Message : Ġ( Message ) Ġ-> ĠMessage Ġ= Ġ{ Ġit Ġ} ĊĊĠ Ġinit Ġ{ ĊĠĠĠ Ġthis . state Ġ= ĠState . CLOSED ĊĠĠĠ Ġthis . bindings Ġ= ĠConcurrent Linked Queue () ĊĠĠĠ Ġthis . binding Ref Ġ= Ġ 0 ĊĠĠĠ Ġthis . timeout Ġ= Ġsocket . timeout ĊĠĠĠ Ġthis . joined Once Ġ= Ġfalse ĊĠĠĠ Ġthis . push Buffer Ġ= Ġmutable ListOf () ĊĠĠĠ Ġthis . state Change Refs Ġ= Ġmutable ListOf () ĊĠĠĠ Ġthis . re join Timer Ġ= ĠTimeout Timer ( ĊĠĠĠĠĠĠĠ Ġdispatch Queue Ġ= Ġsocket . dispatch Queue , ĊĠĠĠĠĠĠĠ Ġtimer Calculation Ġ= Ġsocket . re join After Ms , ĊĠĠĠĠĠĠĠ Ġcallback Ġ= Ġ{ Ġif Ġ( socket . isConnected ) Ġre join () Ġ} ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġ// ĠRespond Ġto Ġsocket Ġevents ĊĠĠĠ Ġthis . socket . onError Ġ{ Ġ_, Ġ_ -> Ġthis . re join Timer . reset () Ġ} ĊĠĠĠĠĠ Ġ. apply Ġ{ Ġstate Change Refs . add ( this ) Ġ} ĊĠĠĠ Ġthis . socket . on Open Ġ{ ĊĠĠĠĠĠ Ġthis . re join Timer . reset () ĊĠĠĠĠĠ Ġif Ġ( this . is Er rored ) Ġ{ Ġthis . re join () Ġ} ĊĠĠĠ Ġ}. apply Ġ{ Ġstate Change Refs . add ( this ) Ġ} ĊĊĊĠĠĠ Ġ// ĠSetup ĠPush Ġto Ġbe Ġsent Ġwhen Ġjoining ĊĠĠĠ Ġthis . join Push Ġ= ĠPush ( ĊĠĠĠĠĠĠĠ Ġchannel Ġ= Ġthis , ĊĠĠĠĠĠĠĠ Ġevent Ġ= ĠEvent . JOIN . value , ĊĠĠĠĠĠĠĠ Ġpayload Ġ= Ġparams , ĊĠĠĠĠĠĠĠ Ġtimeout Ġ= Ġtimeout ) ĊĊĠĠĠ Ġ// ĠPerform Ġonce Ġthe ĠChannel Ġhas Ġjoined ĊĠĠĠ Ġthis . join Push . receive (" ok ") Ġ{ ĊĠĠĠĠĠ Ġ// ĠMark Ġthe ĠChannel Ġas Ġjoined ĊĠĠĠĠĠ Ġthis . state Ġ= ĠState . JOIN ED ĊĊĠĠĠĠĠ Ġ// ĠReset Ġthe Ġtimer , Ġprevent ing Ġit Ġfrom Ġattempting Ġto Ġjoin Ġagain ĊĠĠĠĠĠ Ġthis . re join Timer . reset () ĊĊĠĠĠĠĠ Ġ// ĠSend Ġany Ġbuffered Ġmessages Ġand Ġclear Ġthe Ġbuffer ĊĠĠĠĠĠ Ġthis . push Buffer . forEach Ġ{ Ġit . send () Ġ} ĊĠĠĠĠĠ Ġthis . push Buffer . clear () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠPerform Ġif ĠChannel Ġerrors Ġwhile Ġattempting Ġto Ġjoin ĊĠĠĠ Ġthis . join Push . receive (" error ") Ġ{ ĊĠĠĠĠĠ Ġthis . state Ġ= ĠState . ER RO RED ĊĠĠĠĠĠ Ġif Ġ( this . socket . isConnected ) Ġ{ Ġthis . re join Timer . schedule Timeout () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠPerform Ġif ĠChannel Ġtimed Ġout Ġwhile Ġattempting Ġto Ġjoin ĊĠĠĠ Ġthis . join Push . receive (" timeout ") Ġ{ ĊĠĠĠĠĠ Ġ// ĠLog Ġthe Ġtimeout ĊĠĠĠĠĠ Ġthis . socket . log Items (" Channel : Ġtimeouts Ġ$ topic , Ġ$ join Ref Ġafter Ġ$ timeout Ġms ") ĊĊĠĠĠĠĠ Ġ// ĠSend Ġa ĠPush Ġto Ġthe Ġserver Ġto Ġleave Ġthe ĠChannel ĊĠĠĠĠĠ Ġval Ġleave Push Ġ= ĠPush ( ĊĠĠĠĠĠĠĠĠĠ Ġchannel Ġ= Ġthis , ĊĠĠĠĠĠĠĠĠĠ Ġevent Ġ= ĠEvent . LEAVE . value , ĊĠĠĠĠĠĠĠĠĠ Ġtimeout Ġ= Ġthis . timeout ) ĊĠĠĠĠĠ Ġleave Push . send () ĊĊĠĠĠĠĠ Ġ// ĠMark Ġthe ĠChannel Ġas Ġin Ġan Ġerror Ġand Ġattempt Ġto Ġre join Ġif Ġsocket Ġis Ġconnected ĊĠĠĠĠĠ Ġthis . state Ġ= ĠState . ER RO RED ĊĠĠĠĠĠ Ġthis . join Push . reset () ĊĊĠĠĠĠĠ Ġif Ġ( this . socket . isConnected ) Ġ{ Ġthis . re join Timer . schedule Timeout () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠClean Ġup Ġwhen Ġthe Ġchannel Ġcloses ĊĠĠĠ Ġthis . onClose Ġ{ ĊĠĠĠĠĠ Ġ// ĠReset Ġany Ġtimer Ġthat Ġmay Ġbe Ġon - going ĊĠĠĠĠĠ Ġthis . re join Timer . reset () ĊĊĠĠĠĠĠ Ġ// ĠLog Ġthat Ġthe Ġchannel Ġwas Ġleft ĊĠĠĠĠĠ Ġthis . socket . log Items (" Channel : Ġclose Ġ$ topic Ġ$ join Ref ") ĊĊĠĠĠĠĠ Ġ// ĠMark Ġthe Ġchannel Ġas Ġclosed Ġand Ġremove Ġit Ġfrom Ġthe Ġsocket ĊĠĠĠĠĠ Ġthis . state Ġ= ĠState . CLOSED ĊĠĠĠĠĠ Ġthis . socket . remove ( this ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠHandles Ġan Ġerror , Ġattempts Ġto Ġre join ĊĠĠĠ Ġthis . onError Ġ{ ĊĠĠĠĠĠ Ġ// ĠLog Ġthat Ġthe Ġchannel Ġreceived Ġan Ġerror ĊĠĠĠĠĠ Ġthis . socket . log Items (" Channel : Ġerror Ġ$ topic Ġ${ it . payload }") ĊĊĠĠĠĠĠ Ġ// ĠIf Ġerror Ġwas Ġreceived Ġwhile Ġjoining , Ġthen Ġreset Ġthe ĠPush ĊĠĠĠĠĠ Ġif Ġ( is Join ing ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠMake Ġsure Ġthat Ġthe Ġ" ph x _ join " Ġisn 't Ġbuffered Ġto Ġsend Ġonce Ġthe Ġsocket ĊĠĠĠĠĠĠĠ Ġ// Ġreconnect s . ĠThe Ġchannel Ġwill Ġsend Ġa Ġnew Ġjoin Ġevent Ġwhen Ġthe Ġsocket Ġconnects . ĊĠĠĠĠĠĠĠ Ġthis . join Ref ?. let Ġ{ Ġthis . socket . remove From Send Buffer ( it ) Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠReset Ġthe Ġpush Ġto Ġbe Ġused Ġagain Ġlater ĊĠĠĠĠĠĠĠ Ġthis . join Push . reset () ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġ// ĠMark Ġthe Ġchannel Ġas Ġerro red Ġand Ġattempt Ġto Ġre join Ġif Ġsocket Ġis Ġcurrently Ġconnected ĊĠĠĠĠĠ Ġthis . state Ġ= ĠState . ER RO RED ĊĠĠĠĠĠ Ġif Ġ( socket . isConnected ) Ġ{ Ġthis . re join Timer . schedule Timeout () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠPerform Ġwhen Ġthe Ġjoin Ġreply Ġis Ġreceived ĊĠĠĠ Ġthis . on ( Event . REPLY ) Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠ Ġthis . trigger ( reply Event Name ( message . ref ), Ġmessage . raw Payload , Ġmessage . ref , Ġmessage . join Ref ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ// ĠPublic ĠProperties ĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ/** ĠThe Ġref Ġsent Ġduring Ġthe Ġjoin Ġmessage . Ġ*/ ĊĠ Ġval Ġjoin Ref : ĠString ? Ġget () Ġ= Ġjoin Push . ref ĊĊĠ Ġ/** Ġ@ return ĠTrue Ġif Ġthe ĠChannel Ġcan Ġpush Ġmessages Ġ*/ ĊĠ Ġval Ġcan Push : ĠBoolean ĊĠĠĠ Ġget () Ġ= Ġthis . socket . isConnected Ġ&& Ġthis . is Joined ĊĊĠ Ġ/** Ġ@ return : ĠTrue Ġif Ġthe ĠChannel Ġhas Ġbeen Ġclosed Ġ*/ ĊĠ Ġval Ġis Closed : ĠBoolean ĊĠĠĠ Ġget () Ġ= Ġstate Ġ== ĠState . CLOSED ĊĊĠ Ġ/** Ġ@ return : ĠTrue Ġif Ġthe ĠChannel Ġexperienced Ġan Ġerror Ġ*/ ĊĠ Ġval Ġis Er rored : ĠBoolean ĊĠĠĠ Ġget () Ġ= Ġstate Ġ== ĠState . ER RO RED ĊĊĠ Ġ/** Ġ@ return : ĠTrue Ġif Ġthe Ġchannel Ġhas Ġjoined Ġ*/ ĊĠ Ġval Ġis Joined : ĠBoolean ĊĠĠĠ Ġget () Ġ= Ġstate Ġ== ĠState . JOIN ED ĊĊĠ Ġ/** Ġ@ return : ĠTrue Ġif Ġthe Ġchannel Ġhas Ġrequested Ġto Ġjoin Ġ*/ ĊĠ Ġval Ġis Join ing : ĠBoolean ĊĠĠĠ Ġget () Ġ= Ġstate Ġ== ĠState . JOIN ING ĊĊĠ Ġ/** Ġ@ return : ĠTrue Ġif Ġthe Ġchannel Ġhas Ġrequested Ġto Ġleave Ġ*/ ĊĠ Ġval Ġis Le aving : ĠBoolean ĊĠĠĠ Ġget () Ġ= Ġstate Ġ== ĠState . LE AV ING ĊĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ// ĠPublic ĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġfun Ġjoin ( timeout : ĠLong Ġ= Ġthis . timeout ): ĠPush Ġ{ ĊĠĠĠ Ġ// ĠEnsure Ġthat Ġ`. join ()` Ġis Ġcalled Ġonly Ġonce Ġper ĠChannel Ġinstance ĊĠĠĠ Ġif Ġ( joined Once ) Ġ{ ĊĠĠĠĠĠ Ġthrow ĠIllegalStateException ( ĊĠĠĠĠĠĠĠĠĠ Ġ" Tri ed Ġto Ġjoin Ġchannel Ġmultiple Ġtimes . Ġ` join ()` Ġcan Ġonly Ġbe Ġcalled Ġonce Ġper Ġchannel ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠJoin Ġthe Ġchannel ĊĠĠĠ Ġthis . timeout Ġ= Ġtimeout ĊĠĠĠ Ġthis . joined Once Ġ= Ġtrue ĊĠĠĠ Ġthis . re join () ĊĠĠĠ Ġreturn Ġjoin Push ĊĠ Ġ} ĊĊĠ Ġfun ĠonClose ( callback : Ġ( Message ) Ġ-> ĠUnit ): ĠInt Ġ{ ĊĠĠĠ Ġreturn Ġthis . on ( Event . CLOSE , Ġcallback ) ĊĠ Ġ} ĊĊĠ Ġfun ĠonError ( callback : Ġ( Message ) Ġ-> ĠUnit ): ĠInt Ġ{ ĊĠĠĠ Ġreturn Ġthis . on ( Event . ERROR , Ġcallback ) ĊĠ Ġ} ĊĊĠ Ġfun Ġon Message ( callback : Ġ( Message ) Ġ-> ĠMessage ) Ġ{ ĊĠĠĠ Ġthis . on Message Ġ= Ġcallback ĊĠ Ġ} ĊĊĠ Ġfun Ġon ( event : ĠEvent , Ġcallback : Ġ( Message ) Ġ-> ĠUnit ): ĠInt Ġ{ ĊĠĠĠ Ġreturn Ġthis . on ( event . value , Ġcallback ) ĊĠ Ġ} ĊĊĠ Ġfun Ġon ( event : ĠString , Ġcallback : Ġ( Message ) Ġ-> ĠUnit ): ĠInt Ġ{ ĊĠĠĠ Ġval Ġref Ġ= Ġbinding Ref ĊĠĠĠ Ġthis . binding Ref Ġ= Ġref Ġ+ Ġ 1 ĊĊĠĠĠ Ġthis . bindings . add ( Binding ( event , Ġref , Ġcallback )) ĊĠĠĠ Ġreturn Ġref ĊĠ Ġ} ĊĊĠ Ġfun Ġoff ( event : ĠString , Ġref : ĠInt ? Ġ= Ġnull ) Ġ{ ĊĠĠĠ Ġthis . bindings . removeAll Ġ{ Ġbind Ġ-> ĊĠĠĠĠĠ Ġbind . event Ġ== Ġevent Ġ&& Ġ( ref Ġ== Ġnull Ġ|| Ġref Ġ== Ġbind . ref ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġfun Ġpush ( event : ĠString , Ġpayload : ĠPayload , Ġtimeout : ĠLong Ġ= Ġthis . timeout ): ĠPush Ġ{ ĊĠĠĠ Ġif Ġ(! joined Once ) Ġ{ ĊĠĠĠĠĠ Ġ// ĠIf Ġthe ĠChannel Ġhas Ġnot Ġbeen Ġjoined , Ġthrow Ġan Ġexception ĊĠĠĠĠĠ Ġthrow ĠRuntimeException ( ĊĠĠĠĠĠĠĠĠĠ Ġ" Tri ed Ġto Ġpush Ġ$ event Ġto Ġ$ topic Ġbefore Ġjoining . ĠUse Ġchannel . join () Ġbefore Ġpushing Ġevents ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġpush Event Ġ= ĠPush ( this , Ġevent , Ġpayload , Ġtimeout ) ĊĊĠĠĠ Ġif Ġ( can Push ) Ġ{ ĊĠĠĠĠĠ Ġpush Event . send () ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠ Ġpush Event . start Timeout () ĊĠĠĠĠĠ Ġpush Buffer . add ( push Event ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġreturn Ġpush Event ĊĠ Ġ} ĊĊĠ Ġfun Ġleave ( timeout : ĠLong Ġ= Ġthis . timeout ): ĠPush Ġ{ ĊĠĠĠ Ġ// ĠCan Ġpush Ġis Ġdependent Ġupon Ġstate Ġ== ĠJOIN ED . ĠOnce Ġwe Ġset Ġit Ġto ĠLE AV ING , Ġthen Ġcan Push ĊĠĠĠ Ġ// Ġwill Ġreturn Ġfalse , Ġso Ġinstead Ġstore Ġit Ġ_ before _ Ġstarting Ġthe Ġleave ĊĠĠĠ Ġval Ġcan Push Ġ= Ġthis . can Push ĊĊĠĠĠ Ġ// ĠIf Ġattempting Ġa Ġre join Ġduring Ġa Ġleave , Ġthen Ġreset , Ġcancell ing Ġthe Ġre join ĊĠĠĠ Ġthis . re join Timer . reset () ĊĊĠĠĠ Ġ// ĠPrevent Ġentering Ġa Ġre join Ġloop Ġif Ġleaving Ġa Ġchannel Ġbefore Ġjoined ĊĠĠĠ Ġthis . join Push . cancel Timeout () ĊĊĠĠĠ Ġ// ĠNow Ġset Ġthe Ġstate Ġto Ġleaving ĊĠĠĠ Ġthis . state Ġ= ĠState . LE AV ING ĊĊĠĠĠ Ġ// ĠPerform Ġthe Ġsame Ġbehavior Ġif Ġthe Ġchannel Ġleaves Ġsuccessfully Ġor Ġnot ĊĠĠĠ Ġval ĠonClose : Ġ(( Message ) Ġ-> ĠUnit ) Ġ= Ġ{ ĊĠĠĠĠĠ Ġthis . socket . log Items (" Channel : Ġleave Ġ$ topic ") ĊĠĠĠĠĠ Ġthis . trigger ( Event . CLOSE , Ġmap Of (" reason " Ġto Ġ" leave ")) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠPush Ġevent Ġto Ġsend Ġto Ġthe Ġserver ĊĠĠĠ Ġval Ġleave Push Ġ= ĠPush ( ĊĠĠĠĠĠĠĠ Ġchannel Ġ= Ġthis , ĊĠĠĠĠĠĠĠ Ġevent Ġ= ĠEvent . LEAVE . value , ĊĠĠĠĠĠĠĠ Ġtimeout Ġ= Ġtimeout ) ĊĊĠĠĠ Ġleave Push ĊĠĠĠĠĠĠĠ Ġ. receive (" ok ", ĠonClose ) ĊĠĠĠĠĠĠĠ Ġ. receive (" timeout ", ĠonClose ) ĊĠĠĠ Ġleave Push . send () ĊĊĠĠĠ Ġ// ĠIf Ġthe ĠChannel Ġcannot Ġsend Ġpush Ġevents , Ġtrigger Ġa Ġsuccess Ġlocally ĊĠĠĠ Ġif Ġ(! can Push ) Ġleave Push . trigger (" ok ", Ġhash Map Of ()) ĊĊĠĠĠ Ġreturn Ġleave Push ĊĠ Ġ} ĊĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ// ĠInternal ĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ/** ĠChecks Ġif Ġa ĠMessage 's Ġevent Ġbelongs Ġto Ġthis ĠChannel Ġinstance Ġ*/ ĊĠ Ġinternal Ġfun Ġis Member ( message : ĠMessage ): ĠBoolean Ġ{ ĊĠĠĠ Ġif Ġ( message . topic Ġ!= Ġthis . topic ) Ġreturn Ġfalse ĊĊĠĠĠ Ġval Ġis Lifecycle Event Ġ= ĠEvent . is Lifecycle Event ( message . event ) ĊĊĠĠĠ Ġ// ĠIf Ġthe Ġmessage Ġis Ġa Ġlifecycle Ġevent Ġand Ġit Ġis Ġnot Ġa Ġjoin Ġfor Ġthis Ġchannel , Ġdrop Ġthe Ġoutdated Ġmessage ĊĠĠĠ Ġif Ġ( message . join Ref Ġ!= Ġnull Ġ&& Ġis Lifecycle Event Ġ&& Ġmessage . join Ref Ġ!= Ġthis . join Ref ) Ġ{ ĊĠĠĠĠĠ Ġthis . socket . log Items (" Channel : ĠD ro pping Ġoutdated Ġmessage . Ġ${ message . topic }") ĊĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġreturn Ġtrue ĊĠ Ġ} ĊĊĠ Ġinternal Ġfun Ġtrigger ( ĊĠĠĠ Ġevent : ĠEvent , ĊĠĠĠ Ġpayload : ĠPayload Ġ= Ġhash Map Of (), ĊĠĠĠ Ġref : ĠString Ġ= Ġ"", ĊĠĠĠ Ġjoin Ref : ĠString ? Ġ= Ġnull ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġthis . trigger ( event . value , Ġpayload , Ġref , Ġjoin Ref ) ĊĠ Ġ} ĊĊĠ Ġinternal Ġfun Ġtrigger ( ĊĠĠĠ Ġevent : ĠString , ĊĠĠĠ Ġpayload : ĠPayload Ġ= Ġhash Map Of (), ĊĠĠĠ Ġref : ĠString Ġ= Ġ"", ĊĠĠĠ Ġjoin Ref : ĠString ? Ġ= Ġnull ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġthis . trigger ( Message ( join Ref , Ġref , Ġtopic , Ġevent , Ġpayload )) ĊĠ Ġ} ĊĊĠ Ġinternal Ġfun Ġtrigger ( message : ĠMessage ) Ġ{ ĊĠĠĠ Ġ// ĠInform Ġthe Ġon Message Ġhook Ġof Ġthe Ġmessage ĊĠĠĠ Ġval Ġhandled Message Ġ= Ġthis . on Message ( message ) ĊĊĠĠĠ Ġ// ĠInform Ġall Ġmatching Ġevent Ġbindings Ġof Ġthe Ġmessage ĊĠĠĠ Ġthis . bindings ĊĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . event Ġ== Ġmessage . event Ġ} ĊĠĠĠĠĠĠĠ Ġ. forEach Ġ{ Ġit . callback ( handled Message ) Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĠCreate Ġan Ġevent Ġwith Ġa Ġgiven Ġref Ġ*/ ĊĠ Ġinternal Ġfun Ġreply Event Name ( ref : ĠString ): ĠString Ġ{ ĊĠĠĠ Ġreturn Ġ" chan _ reply _$ ref " ĊĠ Ġ} ĊĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ// ĠPrivate ĊĠ Ġ// ---------------------------------------------------------------- -------------- ĊĠ Ġ/** ĠSends Ġthe ĠChannel 's Ġjoin Push Ġto Ġthe ĠServer Ġ*/ ĊĠ Ġprivate Ġfun Ġsend Join ( timeout : ĠLong ) Ġ{ ĊĠĠĠ Ġthis . state Ġ= ĠState . JOIN ING ĊĠĠĠ Ġthis . join Push . res end ( timeout ) ĊĠ Ġ} ĊĊĠ Ġ/** ĠRe join s Ġthe ĠChannel Ġe . g . Ġafter Ġa Ġdisconnect Ġ*/ ĊĠ Ġprivate Ġfun Ġre join ( timeout : ĠLong Ġ= Ġthis . timeout ) Ġ{ ĊĠĠĠ Ġ// ĠDo Ġnot Ġattempt Ġto Ġre join Ġif Ġthe Ġchannel Ġis Ġin Ġthe Ġprocess Ġof Ġleaving ĊĠĠĠ Ġif Ġ( is Le aving ) Ġreturn ĊĊĠĠĠ Ġ// ĠLeave Ġpotentially Ġduplicated Ġchannels ĊĠĠĠ Ġthis . socket . leave Open Topic ( this . topic ) ĊĊĠĠĠ Ġ// ĠSend Ġthe Ġjoin Push ĊĠĠĠ Ġthis . send Join ( timeout ) ĊĠ Ġ} Ċ }