package Ġnet . cord a . node . services . messaging Ċ Ċ import Ġco . parallel universe . fib ers . Suspend able Ċ import Ġcom . co da h ale . metrics . Metric Registry Ċ import Ġnet . cord a . core . crypto . toString Short Ċ import Ġnet . cord a . core . flows . StateMachine Run Id Ċ import Ġnet . cord a . core . identity . C ord a X 5 0 0 Name Ċ import Ġnet . cord a . core . internal . Named Cache Factory Ċ import Ġnet . cord a . core . internal . Thread Box Ċ import Ġnet . cord a . core . messaging . C ord a RP CO ps Ċ import Ġnet . cord a . core . messaging . Message Recipients Ċ import Ġnet . cord a . core . messaging . Single Message Recipient Ċ import Ġnet . cord a . core . node . NodeInfo Ċ import Ġnet . cord a . core . node . services . Network Map Cache Ċ import Ġnet . cord a . core . node . services . Party Info Ċ import Ġnet . cord a . core . serialization . Serialization Defaults Ċ import Ġnet . cord a . core . serialization . Singleton Serialize As Token Ċ import Ġnet . cord a . core . serialization . deserialize Ċ import Ġnet . cord a . core . serialization . internal . node Serialization Env Ċ import Ġnet . cord a . core . serialization . serialize Ċ import Ġnet . cord a . core . utilities .* Ċ import Ġnet . cord a . node . VersionInfo Ċ import Ġnet . cord a . node . internal . Lifecycle Support Ċ import Ġnet . cord a . node . internal . art emis . Reactive Art emis Consumer Ċ import Ġnet . cord a . node . internal . art emis . Reactive Art emis Consumer . Companion . multip lex Ċ import Ġnet . cord a . node . services . api . Network Map Cache Internal Ċ import Ġnet . cord a . node . services . config . Node Configuration Ċ import Ġnet . cord a . node . services . state machine . Ded up lication Id Ċ import Ġnet . cord a . node . services . state machine . External Event Ċ import Ġnet . cord a . node . services . state machine . Sender Ded up lication Id Ċ import Ġnet . cord a . node . utilities . Affinity Executor Ċ import Ġnet . cord a . node . utilities . error And Terminate Ċ import Ġnet . cord a . node api . internal . Art emis Messaging Component Ċ import Ġnet . cord a . node api . internal . Art emis Messaging Component .* Ċ import Ġnet . cord a . node api . internal . Art emis Messaging Component . Companion . BR ID GE _ CONTROL Ċ import Ġnet . cord a . node api . internal . Art emis Messaging Component . Companion . BR ID GE _ NOTIFY Ċ import Ġnet . cord a . node api . internal . Art emis Messaging Component . Companion . JO URN AL _ HEADER _ SIZE Ċ import Ġnet . cord a . node api . internal . Art emis Messaging Component . Companion . P 2 PM ess aging Headers Ċ import Ġnet . cord a . node api . internal . Art emis Messaging Component . Companion . PE ERS _ PREFIX Ċ import Ġnet . cord a . node api . internal . Art emis Tcp Transport . Companion . p 2 p Connector Tcp Transport Ċ import Ġnet . cord a . node api . internal . brid ging . Bridge Control Ċ import Ġnet . cord a . node api . internal . brid ging . Bridge Entry Ċ import Ġnet . cord a . node api . internal . lifecycle . Service State Helper Ċ import Ġnet . cord a . node api . internal . lifecycle . Service State Support Ċ import Ġnet . cord a . node api . internal . persistence . C ord a Persistence Ċ import Ġnet . cord a . node api . internal . require Message Size Ċ import Ġorg . apache . activ emq . art emis . api . config . Active MQ Default Configuration Ċ import Ġorg . apache . activ emq . art emis . api . core . Active MQ Object Closed Exception Ċ import Ġorg . apache . activ emq . art emis . api . core . Message . HDR _ DUPLIC ATE _ DE TE CTION _ ID Ċ import Ġorg . apache . activ emq . art emis . api . core . Message . HDR _ VALID ATED _ USER Ċ import Ġorg . apache . activ emq . art emis . api . core . Routing Type Ċ import Ġorg . apache . activ emq . art emis . api . core . Simple String Ċ import Ġorg . apache . activ emq . art emis . api . core . client .* Ċ import Ġrx . Observable Ċ import Ġrx . Subscription Ċ import Ġrx . subjects . Publish Subject Ċ import Ġjava . security . PublicKey Ċ import Ġjava . time . Instant Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġjava . util . concurrent . Count DownLatch Ċ import Ġjavax . annotation . concurrent . Thread Safe Ċ import Ġkotlin . concurrent . timer Ċ Ċ /** Ċ Ġ* ĠThis Ġclass Ġimplements Ġthe Ġ[ Messaging Service ] ĠAPI Ġusing ĠApache ĠArt emis , Ġthe Ġsuccess or Ġto Ġtheir ĠActive MQ Ġproduct . Ċ Ġ* ĠArt emis Ġis Ġa Ġmessage Ġqueue Ġbroker Ġand Ġhere Ġwe Ġrun Ġa Ġclient Ġconnecting Ġto Ġthe Ġspecified Ġbroker Ġinstance Ċ Ġ* Ġ[ Art emis Messaging Server ]. ĠIt 's Ġprimarily Ġconcerned Ġwith Ġpeer - to - peer Ġmessaging . Ċ Ġ* Ċ Ġ* ĠMessage Ġhandlers Ġare Ġrun Ġon Ġthe Ġprovided Ġ[ Affinity Executor ] Ġsynchronous ly , Ġthat Ġis , Ġthe ĠArt emis Ġcallback Ġthreads Ċ Ġ* Ġare Ġblocked Ġuntil Ġthe Ġhandler Ġis Ġscheduled Ġand Ġcompleted . ĠThis Ġallows Ġback pressure Ġto Ġpropagate Ġfrom Ġthe Ġgiven Ċ Ġ* Ġexecutor Ġthrough Ġinto ĠArt emis Ġand Ġfrom Ġthere , Ġback Ġthrough Ġto Ġsend ers . Ċ Ġ* Ċ Ġ* ĠAn Ġimplementation Ġof Ġ[ C ord a RP CO ps ] Ġcan Ġbe Ġprovided . ĠIf Ġgiven , Ġclients Ġusing Ġthe ĠC ord a MQ Client ĠRPC Ġlibrary Ġcan Ċ Ġ* Ġinvoke Ġmethods Ġon Ġthe Ġprovided Ġimplementation . ĠThere Ġis Ġmore Ġdocumentation Ġon Ġthis Ġin Ġthe Ġdoc site Ġand Ġthe Ċ Ġ* ĠC ord a RPC Client Ġclass . Ċ Ġ* Ċ Ġ* Ġ@ param Ġconfig ĠThe Ġconfiguration Ġof Ġthe Ġnode , Ġwhich Ġis Ġused Ġfor Ġcontrolling Ġthe Ġmessage Ġre delivery Ġoptions . Ċ Ġ* Ġ@ param Ġversion Info ĠAll Ġmessages Ġfrom Ġthe Ġnode Ġcarry Ġthe Ġversion Ġinfo Ġand Ġreceived Ġmessages Ġare Ġchecked Ġagainst Ġthis Ġfor Ġcompatibility . Ċ Ġ* Ġ@ param Ġserver Address ĠThe Ġhost Ġand Ġport Ġof Ġthe ĠArt emis Ġbroker . Ċ Ġ* Ġ@ param Ġnode Executor ĠThe Ġreceived Ġmessages Ġare Ġmarshal led Ġonto Ġthe Ġserver Ġexecutor Ġto Ġprevent ĠNet ty Ġbuffers Ġle aking Ġduring Ġfiber Ġsus p ends . Ċ Ġ* Ġ@ param Ġdatabase ĠThe Ġnodes Ġdatabase , Ġwhich Ġis Ġused Ġto Ġde duplicate Ġmessages . Ċ Ġ*/ Ċ @ Thread Safe Ċ class ĠP 2 PM ess aging Client ( val Ġconfig : ĠNode Configuration , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġversion Info : ĠVersion Info , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġserver Address : ĠNetwork Host And Port , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġnode Executor : ĠAff inity Executor . Service Affinity Executor , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġdatabase : ĠC ord a Persistence , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġnetwork Map : ĠNetwork Map Cache Internal , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UNUSED ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġmetric Registry : ĠMetric Registry , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcache Factory : ĠNamed Cache Factory , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġis Dr aining Mode On : Ġ() Ġ-> ĠBoolean , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġdr aining Mode Was Changed Events : ĠObservable < Pair < Boolean , ĠBoolean >>, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġstate Helper : ĠService State Helper Ġ= ĠService State Helper ( log ) Ċ ) Ġ: ĠSingleton Serialize As Token (), ĠMessaging Service , ĠAddress To Art emis Queue Resolver , ĠService State Support Ġby Ġstate Helper Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog Ġ= Ġcontext Logger () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġclass ĠNode Client Message ( override Ġval Ġtopic : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġdata : ĠByte Sequence , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġunique MessageId : ĠD ed up lication Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġsender UUID : ĠString ?, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġadditional Headers : ĠMap < String , ĠString >) Ġ: ĠMessage Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġdebug Timestamp : ĠInstant Ġ= ĠInstant . now () ĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠtoString () Ġ= Ġ"$ topic # ${ String ( data . bytes )}" ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġclass ĠInner State Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġstarted Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġvar Ġrunning Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġvar Ġevents Subscription : ĠSubscription ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġp 2 p Consumer : ĠP 2 PM ess aging Consumer ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġlocator : ĠServer Locator ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġexecutor Producer : ĠClient Producer ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġexecutor Session : ĠClient Session ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġproducer : ĠClient Producer ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġproducer Session : ĠClient Session ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġbridge Session : ĠClient Session ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġbridge Notify Consumer : ĠClient Consumer ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġnetwork Change Subscription : ĠSubscription ? Ġ= Ġnull ĊĊĠĠĠĠĠĠĠ Ġfun ĠsendMessage ( address : ĠString , Ġmessage : ĠClient Message ) Ġ= Ġproducer !!. send ( address , Ġmessage ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠA Ġregistration Ġto Ġhandle Ġmessages Ġof Ġdifferent Ġtypes Ġ*/ ĊĠĠĠ Ġdata Ġclass ĠHandler Registration ( val Ġtopic : ĠString , Ġval Ġcallback : ĠAny ) Ġ: ĠMessage Handler Registration ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġmy Identity : ĠPublicKey ĊĠĠĠ Ġprivate Ġvar Ġservice Identity : ĠPublicKey ? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġadvert ised Address : ĠNetwork Host And Port ĊĠĠĠ Ġprivate Ġvar Ġmax Message Size : ĠInt Ġ= Ġ- 1 ĊĊĠĠĠ Ġoverride Ġval Ġmy Address : ĠSingle Message Recipient Ġget () Ġ= ĠNode Address ( my Identity ) ĊĠĠĠ Ġoverride Ġval Ġour Sender UUID Ġ= ĠUUID . randomUUID (). toString () ĊĊĠĠĠ Ġprivate Ġval Ġstate Ġ= ĠThread Box ( Inner State ()) ĊĠĠĠ Ġprivate Ġval Ġknown Queues Ġ= ĠCollections . new Set From Map ( Concurrent HashMap < String , ĠBoolean >()) ĊĠĠĠ Ġprivate Ġval Ġdelay Start Queues Ġ= ĠCollections . new Set From Map ( Concurrent HashMap < String , ĠBoolean >()) ĊĊĠĠĠ Ġprivate Ġval Ġhandlers Ġ= ĠConcurrent HashMap < String , ĠMessage Handler >() ĊĠĠĠ Ġprivate Ġval Ġhandlers Changed Signal Ġ= ĠObject () ĊĊĠĠĠ Ġprivate Ġval Ġde d uplic ator Ġ= ĠP 2 P Message Ded uplic ator ( cache Factory , Ġdatabase ) ĊĠĠĠ Ġinternal Ġvar Ġmessaging Executor : ĠMessaging Executor ? Ġ= Ġnull ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ@ param Ġmy Identity ĠThe Ġprimary Ġidentity Ġof Ġthe Ġnode , Ġwhich Ġdefines Ġthe Ġmessaging Ġaddress Ġfor Ġextern ally Ġreceived Ġmessages . ĊĠĠĠĠ Ġ* ĠIt Ġis Ġalso Ġused Ġto Ġconstruct Ġthe Ġmy Address Ġfield , Ġwhich Ġis Ġultimately Ġadvert ised Ġin Ġthe Ġnetwork Ġmap . ĊĠĠĠĠ Ġ* Ġ@ param Ġservice Identity ĠAn Ġoptional Ġsecond Ġidentity Ġif Ġthe Ġnode Ġis Ġalso Ġpart Ġof Ġa Ġgroup Ġaddress , Ġfor Ġexample Ġa Ġnot ary . ĊĠĠĠĠ Ġ* Ġ@ param Ġadvert ised Address ĠThe Ġextern ally Ġadvert ised Ġversion Ġof Ġthe ĠArt emis Ġbroker Ġaddress Ġused Ġto Ġconstruct Ġmy Address Ġand Ġincluded ĊĠĠĠĠ Ġ* Ġin Ġthe Ġnetwork Ġmap Ġdata . ĊĠĠĠĠ Ġ* Ġ@ param Ġmax Message Size ĠA Ġbound Ġapplied Ġto Ġthe Ġmessage Ġsize . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġstart ( my Identity : ĠPublicKey , Ġservice Identity : ĠPublicKey ?, Ġmax Message Size : ĠInt , Ġadvert ised Address : ĠNetwork Host And Port Ġ= Ġserver Address ) Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . my Identity Ġ= Ġmy Identity ĊĠĠĠĠĠĠĠ Ġthis . service Identity Ġ= Ġservice Identity ĊĠĠĠĠĠĠĠ Ġthis . advert ised Address Ġ= Ġadvert ised Address ĊĠĠĠĠĠĠĠ Ġthis . max Message Size Ġ= Ġmax Message Size ĊĠĠĠĠĠĠĠ Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstarted Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Connecting Ġto Ġmessage Ġbroker : Ġ$ server Address ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO ĠAdd Ġbroker ĠCN Ġto Ġconfig Ġfor Ġhost Ġverification Ġin Ġcase Ġthe Ġembedded Ġbroker Ġisn 't Ġused ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtcp Transport Ġ= Ġp 2 p Connector Tcp Transport ( server Address , Ġconfig . p 2 p Ssl Options ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlocator Ġ= ĠActive MQ Client . create Server Locator Without HA ( tcp Transport ). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNever Ġtime Ġout Ġon Ġour Ġloop back ĠArt emis Ġconnections . ĠIf Ġwe Ġswitch Ġback Ġto Ġusing Ġthe ĠIn VM Ġtransport Ġthis ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġwould Ġbe Ġthe Ġdefault Ġand Ġthe Ġtwo Ġlines Ġbelow Ġcan Ġbe Ġdeleted . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection TTL Ġ= Ġ 6 0 0 0 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Failure Check Period Ġ= Ġ 3 0 0 0 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmin Large Message Size Ġ= Ġmax Message Size Ġ+ ĠJ O URN AL _ HEADER _ SIZE ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis Use Global Pools Ġ= Ġnode Serialization Env Ġ!= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsession Factory Ġ= Ġlocator !!. create SessionFactory (). add F ailover Listener (:: fa ilover Callback ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠLogin Ġusing Ġthe Ġnode Ġusername . ĠThe Ġbroker Ġwill Ġauthenticate Ġus Ġas Ġits Ġnode Ġ( as Ġopposed Ġto Ġanother Ġpeer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġusing Ġour ĠTLS Ġcertificate . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNote Ġthat Ġthe Ġacknowledge ment Ġof Ġmessages Ġis Ġnot Ġfl ushed Ġto Ġthe ĠAr ter mis Ġjournal Ġuntil Ġthe Ġdefault Ġbuffer ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsize Ġof Ġ 1 MB Ġis Ġacknowled ged . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcreate New Session Ġ= Ġ{ Ġsession Factory !!. create Session ( Art emis Messaging Component . NODE _ P 2 P _ USER , ĠArt emis Messaging Component . NODE _ P 2 P _ USER , Ġfalse , Ġtrue , Ġtrue , Ġfalse , ĠActive MQ Client . DEFAULT _ ACK _ BATCH _ SIZE ) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor Session Ġ= Ġcreate New Session () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer Session Ġ= Ġcreate New Session () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbridge Session Ġ= Ġcreate New Session () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor Session !!. start () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer Session !!. start () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbridge Session !!. start () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġin boxes Ġ= Ġmutable Set Of < String >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCreate Ġa Ġqueue , Ġconsumer Ġand Ġproducer Ġfor Ġhandling ĠP 2 P Ġnetwork Ġmessages . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCreate Ġa Ġgeneral Ġpurpose Ġproducer . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer Ġ= Ġproducer Session !!. create Producer () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor Producer Ġ= Ġexecutor Session !!. create Producer () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġin boxes Ġ+= ĠRemote Inbox Address ( my Identity ). queue Name ĊĠĠĠĠĠĠĠĠĠĠĠ Ġservice Identity ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġin boxes Ġ+= ĠRemote Inbox Address ( it ). queue Name ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġin boxes . forEach Ġ{ Ġcreate Queue If Absent ( it , Ġproducer Session !! , Ġexclusive Ġ= Ġtrue ) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġp 2 p Consumer Ġ= ĠP 2 PM ess aging Consumer ( in boxes , Ġcreate New Session , Ġis Dr aining Mode On , Ġdr aining Mode Was Changed Events , Ġmetric Registry ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessaging Executor Ġ= ĠMessaging Executor ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor Session !! , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor Producer !! , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġversion Info , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis @ P 2 PM ess aging Client , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġour Sender UUID Ġ= Ġour Sender UUID ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġregister Bridge Control ( bridge Session !! , Ġin boxes . toList ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġenumerate B rid ges ( bridge Session !! , Ġin boxes . toList ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġfailover Callback ( event : ĠF ailover EventType ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ( event ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠF ailover EventType . FAILURE _ DETECT ED Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror And Terminate (" Connection Ġto Ġthe Ġbroker Ġwas Ġlost . ĠNode Ġis Ġshut ting Ġdown .", Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠF ailover EventType . FA I LO VER _ FAILED Ġ-> Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( running ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror And Terminate (" Could Ġnot Ġreconnect Ġto Ġthe Ġbroker . ĠNode Ġis Ġshut ting Ġdown .", Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Cannot Ġhandle Ġevent Ġ$ event .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠInner State . register Bridge Control ( session : ĠClient Session , Ġin boxes : ĠList < String >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbridge Notify Queue Ġ= Ġ"$ BR ID GE _ NOTIFY .${ my Identity . toString Short () }" ĊĠĠĠĠĠĠĠ Ġif Ġ(! session . queue Query ( Simple String ( bridge Notify Queue )). is Exists ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsession . create Temporary Queue ( BR ID GE _ NOTIFY , ĠRouting Type . MULT IC AST , Ġbridge Notify Queue ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġbridge Consumer Ġ= Ġsession . create Consumer ( bridge Notify Queue ) ĊĠĠĠĠĠĠĠ Ġbridge Notify Consumer Ġ= Ġbridge Consumer ĊĠĠĠĠĠĠĠ Ġbridge Consumer . setMessage Handler Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdata : ĠByteArray Ġ= ĠByteArray ( msg . body Size ). apply Ġ{ Ġmsg . body Buffer . read Bytes ( this ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnotify Message Ġ= Ġdata . deserialize < Bridge Control >( context Ġ= ĠSerialization Defaults . P 2 P _ CONTEXT ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info ( notify Message . toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( notify Message ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠBridge Control . Bridge To Node Snapshot Request Ġ-> Ġenumerate B rid ges ( session , Ġin boxes ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġlog . error (" Unexpected ĠBridge ĠControl Ġmessage Ġtype Ġon Ġnotify Ġtopic Ġ$ notify Message ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . ack nowledge () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġnetwork Change Subscription Ġ= Ġnetwork Map . changed . subscribe Ġ{ Ġupdate B rid ges On Network Change ( it ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġsend Bridge Control ( message : ĠBridge Control ) Ġ{ ĊĠĠĠĠĠĠĠ Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcontrol Packet Ġ= Ġmessage . serialize ( context Ġ= ĠSerialization Defaults . P 2 P _ CONTEXT ). bytes ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġart emis Message Ġ= Ġproducer Session !!. create Message ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġart emis Message . write Body Buffer Bytes ( control Packet ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠsendMessage ( BR ID GE _ CONTROL , Ġart emis Message ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġupdate B rid ges On Network Change ( change : ĠNetwork Map Cache . Map Change ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . info (" Updating Ġb rid ges Ġon Ġnetwork Ġmap Ġchange : Ġ${ change . node }") ĊĠĠĠĠĠĠĠ Ġfun Ġgather Addresses ( node : ĠNode Info ): ĠSequence < Bridge Entry > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnode . legal Id entities And Certs . map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessaging Address Ġ= ĠNode Address ( it . party . own ing Key ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠBridge Entry ( messaging Address . queue Name , Ġnode . addresses , Ġnode . legal Id entities . map Ġ{ Ġit . name Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. filter Ġ{ Ġproducer Session !!. queue Query ( Simple String ( it . queue Name )). is Exists Ġ}. as Sequence () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġdeploy B rid ges ( node : ĠNode Info ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġgather Addresses ( node ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Bridge Control ( Bridge Control . Create ( config . my Legal Name . toString (), Ġit )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġdestroy B rid ges ( node : ĠNode Info ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġgather Addresses ( node ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Bridge Control ( Bridge Control . Delete ( config . my Legal Name . toString (), Ġit )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġwhen Ġ( change ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠNetwork Map Cache . Map Change . Added Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy B rid ges ( change . node ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠNetwork Map Cache . Map Change . Removed Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdestroy B rid ges ( change . node ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠNetwork Map Cache . Map Change . Modified Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdestroy B rid ges ( change . previous Node ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy B rid ges ( change . node ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġenumerate B rid ges ( session : ĠClient Session , Ġin boxes : ĠList < String >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrequired B rid ges Ġ= Ġmutable ListOf < Bridge Entry >() ĊĠĠĠĠĠĠĠ Ġfun Ġcreate Bridge Entry ( queue Name : ĠSimple String ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Hash Ġ= Ġqueue Name . substring ( PE ERS _ PREFIX . length ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpeers Ġ= Ġnetwork Map . get Nodes By Ow ning Key Index ( key Hash ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ( node Ġin Ġpeers ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbridge Ġ= ĠBridge Entry ( queue Name . toString (), Ġnode . addresses , Ġnode . legal Id entities . map Ġ{ Ġit . name Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequired B rid ges Ġ+= Ġbridge ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġknown Queues Ġ+= Ġqueue Name . toString () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġqueues Ġ= Ġsession . address Query ( Simple String ("$ PE ERS _ PREFIX # ")). queue Names ĊĠĠĠĠĠĠĠ Ġknown Queues . clear () ĊĠĠĠĠĠĠĠ Ġfor Ġ( queue Ġin Ġqueues ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġqueue Query Ġ= Ġsession . queue Query ( queue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! config . lazy Bridge Start Ġ|| Ġqueue Query . message Count Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Bridge Entry ( queue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay Start Queues Ġ+= Ġqueue . toString () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġstartup Message Ġ= ĠBridge Control . Node To Bridge Snapshot ( config . my Legal Name . toString (), Ġin boxes , Ġrequired B rid ges ) ĊĠĠĠĠĠĠĠ Ġsend Bridge Control ( startup Message ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġshutdown Latch Ġ= ĠCount DownLatch ( 1 ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠStarts Ġthe Ġp 2 p Ġevent Ġloop : Ġthis Ġmethod Ġonly Ġreturns Ġonce Ġ[ stop ] Ġhas Ġbeen Ġcalled . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġoverride Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġl atch Ġ= ĠCount DownLatch ( 1 ) ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( handlers Changed Signal ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( handlers . isEmpty () Ġ&& Ġstate . locked Ġ{ Ġ( p 2 p Consumer Ġ!= Ġnull ) Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhandlers Changed Signal . wait () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck ( started ) Ġ{ Ġ" start Ġmust Ġbe Ġcalled Ġfirst " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck (! running ) Ġ{ Ġ" run Ġcan 't Ġbe Ġcalled Ġtwice " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrunning Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġit 's Ġnull , Ġit Ġmeans Ġwe Ġalready Ġcalled Ġstop , Ġso Ġreturn Ġimmediately . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( p 2 p Consumer Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subscription Ġ= Ġp 2 p Consumer !!. messages ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġ` run ()` Ġmethod Ġis Ġsemantic ally Ġmeant Ġto Ġblock Ġuntil Ġthe Ġmessage Ġconsumption Ġruns , Ġhence Ġthe Ġl atch Ġhere ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Completed ( latch :: count Down ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ({ Ġmessage Ġ-> Ġdeliver ( message ) Ġ}, Ġ{ Ġerror Ġ-> Ġthrow Ġerror Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġp 2 p Consumer !! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . start () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug (" Sign all ing Ġactive ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstate Helper . active Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug (" Await ing Ġon Ġl atch ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġl atch . await () ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshutdown Latch . count Down () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġart emis To C ord a Message ( message : ĠClient Message ): ĠReceived Message ? Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequire Message Size ( message . body Size , Ġmax Message Size ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= Ġmessage . required ( P 2 PM ess aging Headers . topic Property ) Ġ{ ĠgetString Property ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġuser Ġ= Ġrequire NotNull ( message . getString Property ( HDR _ VALID ATED _ USER )) Ġ{ Ġ" Message Ġis Ġnot Ġauthenticated " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġplatform Version Ġ= Ġmessage . required ( P 2 PM ess aging Headers . platform Version Property ) Ġ{ Ġget Int Property ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠUse Ġthe Ġmagic Ġde dup lication Ġproperty Ġbuilt Ġinto ĠArt emis Ġas Ġour Ġmessage Ġidentity Ġtoo ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġunique MessageId Ġ= Ġmessage . required ( HDR _ DUPLIC ATE _ DE TE CTION _ ID ) Ġ{ ĠD ed up lication Id ( message . getString Property ( it )) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreceived Sender UUID Ġ= Ġmessage . getString Property ( P 2 PM ess aging Headers . sender UUID ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreceived Sender Seq No Ġ= Ġif Ġ( message . contains Property ( P 2 PM ess aging Headers . sender Seq No )) Ġmessage . getLong Property ( P 2 PM ess aging Headers . sender Seq No ) Ġelse Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġis Session Init Ġ= Ġmessage . getString Property ( P 2 PM ess aging Headers . Type . KEY ) Ġ== ĠP 2 PM ess aging Headers . Type . SESSION _ INIT _ VALUE ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . trace Ġ{ Ġ" Received Ġmessage Ġfrom : Ġ${ message . address } Ġuser : Ġ$ user Ġtopic : Ġ$ topic Ġid : Ġ$ unique MessageId Ġsender UUID : Ġ$ received Sender UUID Ġsender Seq No : Ġ$ received Sender Seq No Ġis Session Init : Ġ$ is Session Init " Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠArt emis Received Message ( topic , ĠC ord a X 5 0 0 Name . parse ( user ), Ġplatform Version , Ġunique MessageId , Ġreceived Sender UUID , Ġreceived Sender Seq No , Ġis Session Init , Ġmessage ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Unable Ġto Ġprocess Ġmessage , Ġignoring Ġit : Ġ$ message ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġinline Ġfun Ġ< T > ĠClient Message . required ( key : ĠSimple String , Ġextractor : ĠClient Message .( Simple String ) Ġ-> ĠT ): ĠT Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire ( contains Property ( key )) Ġ{ Ġ" Missing Ġ$ key " Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġextractor ( key ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġclass ĠArt emis Received Message ( override Ġval Ġtopic : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġpeer : ĠC ord a X 5 0 0 Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġplatform Version : ĠInt , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġunique MessageId : ĠD ed up lication Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġsender UUID : ĠString ?, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġsender Seq No : ĠLong ?, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġis Session Init : ĠBoolean , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġmessage : ĠClient Message ) Ġ: ĠReceived Message Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġdata : ĠByte Sequence Ġby Ġlazy Ġ{ ĠO pa que Bytes ( ByteArray ( message . body Size ). apply Ġ{ Ġmessage . body Buffer . read Bytes ( this ) Ġ}) Ġ} ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġdebug Timestamp : ĠInstant Ġget () Ġ= ĠInstant . of Epoch Milli ( message . timestamp ) ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġadditional Headers : ĠMap < String , ĠString > Ġ= Ġempty Map () ĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠtoString () Ġ= Ġ"$ topic #$ data " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġdeliver ( art emis Message : ĠClient Message ) Ġ{ ĊĠĠĠĠĠĠĠ Ġart emis To C ord a Message ( art emis Message )?. let Ġ{ Ġc ord a Message Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! ded uplic ator . is Duplicate ( cord a Message )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġde d uplic ator . signal Message Process Start ( cord a Message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdeliver ( cord a Message , Ġart emis Message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . trace Ġ{ Ġ" Discard Ġduplicate Ġmessage Ġ${ cord a Message . unique MessageId } Ġfor Ġ${ cord a Message . topic }" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessaging Executor !!. ack nowledge ( art emis Message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġdeliver ( msg : ĠReceived Message , Ġart emis Message : ĠClient Message ) Ġ{ ĊĠĠĠĠĠĠĠ Ġstate . check Not Locked () ĊĠĠĠĠĠĠĠ Ġval Ġdeliver To Ġ= Ġhandlers [ msg . topic ] ĊĠĠĠĠĠĠĠ Ġif Ġ( deliver To Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdeliver To ( msg , ĠHandler Registration ( msg . topic , Ġdeliver To ), ĠMessage Ded up lication Handler ( art emis Message , Ġmsg )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Ca ught Ġexception Ġwh il st Ġexecuting Ġmessage Ġhandler Ġfor Ġ${ msg . topic }", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Received Ġmessage Ġ${ msg . unique MessageId } Ġfor Ġ${ msg . topic } Ġthat Ġdoesn 't Ġhave Ġany Ġregistered Ġhandlers Ġyet ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġinner Ġclass ĠMessage Ded up lication Handler ( val Ġart emis Message : ĠClient Message , Ġoverride Ġval Ġreceived Message : ĠReceived Message ) Ġ: ĠD ed up lication Handler , ĠExternal Event . External Message Event Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġexternal Cause : ĠExternal Event ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget () Ġ= Ġthis ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġflow Id : ĠState Machine Run Id Ġby Ġlazy Ġ{ ĠState Machine Run Id . create Random () Ġ} ĊĠĠĠĠĠĠĠ Ġoverride Ġval Ġde dup lication Handler : ĠMessage Ded up lication Handler ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget () Ġ= Ġthis ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġinside Database Transaction () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġde d uplic ator . persist Ded up lication Id ( received Message . unique MessageId ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġafter Database Transaction () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġde d uplic ator . signal Message Process Finish ( received Message . unique MessageId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessaging Executor !!. ack nowledge ( art emis Message ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠtoString (): ĠString Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġ"${ java Class . simple Name } (${ received Message . unique MessageId })" ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠInit i ates Ġshutdown : Ġif Ġcalled Ġfrom Ġa Ġthread Ġthat Ġisn 't Ġcontrolled Ġby Ġthe Ġexecutor Ġpassed Ġto Ġthe Ġconstructor ĊĠĠĠĠ Ġ* Ġthen Ġthis Ġwill Ġblock Ġuntil Ġall Ġin - flight Ġmessages Ġhave Ġfinished Ġbeing Ġhandled Ġand Ġacknowled ged . ĠIf Ġcalled ĊĠĠĠĠ Ġ* Ġfrom Ġa Ġthread Ġthat 's Ġa Ġpart Ġof Ġthe Ġ[ net . cord a . node . utilities . Affinity Executor ] Ġgiven Ġto Ġthe Ġconstructor , ĊĠĠĠĠ Ġ* Ġit Ġreturns Ġimmediately Ġand Ġshutdown Ġis Ġasynchronous . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġoverride Ġfun Ġstop () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrunning Ġ= Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġallow Ġstop () Ġto Ġbe Ġcalled Ġwithout Ġa Ġrun () Ġin Ġbetween , Ġbut Ġit Ġmust Ġhave Ġat Ġleast Ġbeen Ġstarted . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheck ( started ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġprev Running Ġ= Ġrunning ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrunning Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstate Helper . active Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnetwork Change Subscription ?. unsubscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( p 2 p Consumer Ġ!= Ġnull , Ġ{ Ġ" stop Ġcan 't Ġbe Ġcalled Ġtwice " Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( producer Ġ!= Ġnull , Ġ{ Ġ" stop Ġcan 't Ġbe Ġcalled Ġtwice " Ġ}) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġclose ( p 2 p Consumer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġp 2 p Consumer Ġ= Ġnull ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġclose ( producer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer Session !!. commit () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġclose ( executor Producer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor Producer Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor Session !!. commit () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġclose ( bridge Notify Consumer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġknown Queues . clear () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subscription ?. unsubscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevents Subscription Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprev Running ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġsynchronized ( handlers Changed Signal ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhandlers Changed Signal . notify All () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( running Ġ&& Ġ! node Executor . is On Thread ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWait Ġfor Ġthe Ġmain Ġloop Ġto Ġnotice Ġthe Ġconsumer Ġhas Ġgone Ġand Ġfinish Ġup . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshutdown Latch . await () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠOnly Ġfirst Ġcaller Ġto Ġgets Ġrunning Ġtrue Ġto Ġprotect Ġagainst Ġdouble Ġstop , Ġwhich Ġseems Ġto Ġhappen Ġin Ġsome Ġintegration Ġtests . ĊĠĠĠĠĠĠĠ Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlocator ?. close () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġclose ( target : ĠAuto Closeable ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtarget ?. close () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ignored : ĠActive MQ Object Closed Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġsw allow ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suspend able ĊĠĠĠ Ġoverride Ġfun Ġsend ( message : ĠMessage , Ġtarget : ĠMessage Recipients , Ġsequence Key : ĠAny ) Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire Message Size ( message . data . size , Ġmax Message Size ) ĊĠĠĠĠĠĠĠ Ġmessaging Executor !!. send ( message , Ġtarget ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suspend able ĊĠĠĠ Ġoverride Ġfun Ġsend ( add ressed Messages : ĠList < Messaging Service . Add ressed Message >) Ġ{ ĊĠĠĠĠĠĠĠ Ġfor Ġ(( message , Ġtarget , Ġsequence Key ) Ġin Ġaddressed Messages ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend ( message , Ġtarget , Ġsequence Key ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġresolve Target To Art emis Queue ( address : ĠMessage Recipients ): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( address Ġ== Ġmy Address ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġwe Ġare Ġsending Ġto Ġourselves Ġthen Ġroute Ġthe Ġmessage Ġdirectly Ġto Ġour ĠP 2 P Ġqueue . ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRemote Inbox Address ( my Identity ). queue Name ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠOtherwise Ġwe Ġsend Ġthe Ġmessage Ġto Ġan Ġinternal Ġqueue Ġfor Ġthe Ġtarget Ġres iding Ġon Ġour Ġbroker . ĠIt 's Ġthen Ġthe ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġbroker 's Ġjob Ġto Ġroute Ġthe Ġmessage Ġto Ġthe Ġtarget 's ĠP 2 P Ġqueue . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinternal Target Queue Ġ= Ġ( address Ġas ? ĠArt emis Address )?. queue Name ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠIllegalArgumentException (" Not Ġan ĠArt emis Ġaddress ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstate . locked Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Queue If Absent ( internal Target Queue , Ġproducer Session !! , Ġexclusive Ġ= Ġaddress Ġ! is ĠService Address ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinternal Target Queue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠAt tempts Ġto Ġcreate Ġa Ġdur able Ġqueue Ġon Ġthe Ġbroker Ġwhich Ġis Ġbound Ġto Ġan Ġaddress Ġof Ġthe Ġsame Ġname . Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġcreate Queue If Absent ( queue Name : ĠString , Ġsession : ĠClient Session , Ġexclusive : ĠBoolean ) Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġsend Bridge Create Message () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Hash Ġ= Ġqueue Name . substring ( PE ERS _ PREFIX . length ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpeers Ġ= Ġnetwork Map . get Nodes By Ow ning Key Index ( key Hash ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ( node Ġin Ġpeers ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbridge Ġ= ĠBridge Entry ( queue Name , Ġnode . addresses , Ġnode . legal Id entities . map Ġ{ Ġit . name Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcreate Bridge Message Ġ= ĠBridge Control . Create ( config . my Legal Name . toString (), Ġbridge ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Bridge Control ( create Bridge Message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ(! known Queues . contains ( queue Name )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( delay Start Queues . contains ( queue Name )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Start Ġbridge Ġfor Ġpreviously Ġempty Ġqueue Ġ$ queue Name ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Bridge Create Message () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay Start Queues Ġ-= Ġqueue Name ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġqueue Query Ġ= Ġsession . queue Query ( Simple String ( queue Name )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! queue Query . is Exists ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Create Ġfresh Ġqueue Ġ$ queue Name Ġbound Ġon Ġsame Ġaddress ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsession . create Queue ( queue Name , ĠRouting Type . ANY CAST , Ġqueue Name , Ġnull , Ġtrue , Ġfalse , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠActive MQ Default Configuration . getDefault Max Queue Consum ers (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠActive MQ Default Configuration . getDefault Purge On No Consum ers (), Ġexclusive , Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend Bridge Create Message () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġknown Queues Ġ+= Ġqueue Name ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġadd MessageHandler ( topic : ĠString , Ġcallback : ĠMessage Handler ): ĠMessage Handler Registration Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire (! topic . isBlank ()) Ġ{ Ġ" Topic Ġmust Ġnot Ġbe Ġblank , Ġas Ġthe Ġempty Ġtopic Ġis Ġa Ġspecial Ġcase ." Ġ} ĊĠĠĠĠĠĠĠ Ġhandlers . compute ( topic ) Ġ{ Ġ_, Ġhandler Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( handler Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Cannot Ġadd Ġanother Ġack ing Ġhandler Ġfor Ġ$ topic , Ġthere Ġis Ġalready Ġan Ġack ing Ġone ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcallback ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġsynchronized ( handlers Changed Signal ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhandlers Changed Signal . notify All () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠHandler Registration ( topic , Ġcallback ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġremove MessageHandler ( registration : ĠMessage Handler Registration ) Ġ{ ĊĠĠĠĠĠĠĠ Ġregistration Ġas ĠHandler Registration ĊĠĠĠĠĠĠĠ Ġhandlers . remove ( registration . topic ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġcreate Message ( topic : ĠString , Ġdata : ĠByteArray , Ġde dup lication Id : ĠSender Ded up lication Id , Ġadditional Headers : ĠMap < String , ĠString >): ĠMessage Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠNode Client Message ( topic , ĠO pa que Bytes ( data ), Ġde dup lication Id . ded up lication Id , Ġde dup lication Id . sender UUID , Ġadditional Headers ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Address Of Party ( party Info : ĠParty Info ): ĠMessage Recipients Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( party Info ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠParty Info . Single Node Ġ-> ĠNode Address ( party Info . party . own ing Key ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠParty Info . Distributed Node Ġ-> ĠService Address ( party Info . party . own ing Key ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġclass ĠP 2 PM ess aging Consumer ( ĊĠĠĠĠĠĠĠ Ġqueue Names : ĠSet < String >, ĊĠĠĠĠĠĠĠ Ġcreate Session : Ġ() Ġ-> ĠClient Session , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġis Dr aining Mode On : Ġ() Ġ-> ĠBoolean , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġdr aining Mode Was Changed Events : ĠObservable < Pair < Boolean , ĠBoolean >>, ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġmetrics Registry Ġ: ĠMetric Registry ) Ġ: ĠLifecycle Support Ġ{ ĊĊĠĠĠ Ġprivate Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval Ġinitial Session Messages Ġ= Ġ"${ P 2 PM ess aging Headers . Type . KEY } < >' ${ P 2 PM ess aging Headers . Type . SESSION _ INIT _ VALUE } '" ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġby Ġlazy Ġ{ Ġlogger For < P 2 PM ess aging Client >() Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġvar Ġstarted Flag Ġ= Ġfalse ĊĊĠĠĠ Ġval Ġmessages : ĠPublish Subject < Client Message > Ġ= ĠPublish Subject . create < Client Message >() ĊĊĠĠĠ Ġprivate Ġval Ġexisting Only Consumer Ġ= Ġmultip lex ( queue Names , Ġcreate Session , Ġinitial Session Messages ) ĊĠĠĠ Ġprivate Ġval Ġinitial And Existing Consumer Ġ= Ġmultip lex ( queue Names , Ġcreate Session ) ĊĠĠĠ Ġprivate Ġval Ġsubscriptions Ġ= Ġmutable Set Of < Subscription >() ĊĊĠĠĠ Ġprivate Ġvar Ġnotification Timer Ġ: ĠTimer ? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġfun Ġschedule Dr ain Notification Timer () Ġ{ ĊĠĠĠĠĠĠĠ Ġnotification Timer Ġ= Ġ Ġtimer (" Dr ain Notification Timer ", Ġtrue , Ġ 1 0 . seconds . to Millis (), Ġ 1 . minutes . to Millis ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Node Ġis Ġcurrently Ġin Ġdr aining Ġmode , Ġnew Ġflows Ġwill Ġnot Ġbe Ġprocessed ! ĠFlow s Ġin Ġflight : Ġ${ metrics Registry . ga u ges [" Flows . In Flight "] ?. value }") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstart () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequire (! started Flag ){ " Must Ġnot Ġalready Ġbe Ġstarted "} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdr aining Mode Was Changed Events . filter Ġ{ Ġchange Ġ-> Ġchange . sw itched On () Ġ}. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinitial And Existing Consumer . switch To ( existing Only Consumer ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġschedule Dr ain Notification Timer () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. subscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdr aining Mode Was Changed Events . filter Ġ{ Ġchange Ġ-> Ġchange . sw itched Off () Ġ}. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexisting Only Consumer . switch To ( initial And Existing Consumer ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnotification Timer ?. cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. subscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriptions Ġ+= Ġexisting Only Consumer . messages . do On Next ( messages :: onNext ). subscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriptions Ġ+= Ġinitial And Existing Consumer . messages . do On Next ( messages :: onNext ). subscribe () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( is Dr aining Mode On ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexisting Only Consumer . start () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġschedule Dr ain Notification Timer () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinitial And Existing Consumer . start () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstarted Flag Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstop () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( started Flag ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexisting Only Consumer . stop () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinitial And Existing Consumer . stop () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriptions . forEach ( Subscription :: unsubscribe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriptions . clear () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstarted Flag Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessages . on Completed () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġval Ġstarted : ĠBoolean ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġstarted Flag ĊĊĠĠĠ Ġprivate Ġfun ĠPair < Boolean , ĠBoolean >. sw itched Off () Ġ= Ġfirst Ġ&& Ġ! second ĊĊĠĠĠ Ġprivate Ġfun ĠPair < Boolean , ĠBoolean >. sw itched On () Ġ= Ġ! first Ġ&& Ġsecond Ċ } Ċ Ċ private Ġfun ĠReactive Art emis Consumer . switch To ( other : ĠReactive Art emis Consumer ) Ġ{ ĊĊĠĠĠ Ġdisconnect () ĊĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠ Ġ! other . started Ġ-> Ġother . start () ĊĠĠĠĠĠĠĠ Ġ! other . connected Ġ-> Ġother . connect () ĊĠĠĠ Ġ} Ċ } Ċ