package Ġch . source motion . vertx . redis . client . heim d all . impl . subscription . connection Ċ Ċ import Ġch . source motion . vertx . redis . client . heim d all . Redis He im d all Exception Ċ import Ġch . source motion . vertx . redis . client . heim d all . Redis He im d all Exception . Reason Ċ import Ġch . source motion . vertx . redis . client . heim d all . impl . connection . Redis He im d all Connection Ċ import Ġch . source motion . vertx . redis . client . heim d all . impl . subscription . Subscription Store Ċ import Ġch . source motion . vertx . redis . client . heim d all . impl . subscription . verify Subscription Command Only Ċ import Ġch . source motion . vertx . redis . client . heim d all . impl . subscription . verify Subscription Commands Only Ċ import Ġio . vertx . core . Composite Future Ċ import Ġio . vertx . core . Future Ċ import Ġio . vertx . core . Handler Ċ import Ġio . vertx . core . Promise Ċ import Ġio . vertx . core . logging . LoggerFactory Ċ import Ġio . vertx . redis . client . Command Ċ import Ġio . vertx . redis . client . Redis Connection Ċ import Ġio . vertx . redis . client . Request Ċ import Ġio . vertx . redis . client . Response Ċ Ċ /** Ċ Ġ* ĠRedis Ġconnection Ġdesigned Ġto Ġbe Ġused Ġfor Ġsubscription Ġpurposes Ġonly . Ċ Ġ*/ Ċ internal Ġclass ĠRedis He im d all Subscription Connection ( ĊĠĠĠ Ġdelegate : ĠRedis Connection , ĊĠĠĠ Ġconnection Issue Handler : ĠHandler < Throwable >, ĊĠĠĠ Ġprivate Ġval Ġsubscription Store : ĠSubscription Store , ĊĠĠĠ Ġprivate Ġval Ġmessage Handler : ĠHandler < Response >, Ċ ) Ġ: ĠRedis He im d all Connection ( delegate , Ġconnection Issue Handler ) Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( Redis He im d all Subscription Connection :: class . java ) ĊĠĠĠĠĠĠĠ Ġval Ġsupported Commands Ġ= ĠlistOf ( Command . SUB SCR IBE , ĠCommand . UN SUB SCR IBE , ĠCommand . PS UB SCR IBE , ĠCommand . PUN SUB SCR IBE ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġinit Connection (): ĠRedis He im d all Connection Ġ{ ĊĠĠĠĠĠĠĠ Ġsuper . init Connection () ĊĠĠĠĠĠĠĠ Ġ// ĠCall Ġsuper Ġinstead Ġof Ġthis Ġreference Ġis Ġmandatory Ġhere , Ġas Ġthis Ġimplementation Ġprevents Ġsetting Ġthe Ġhandler Ġby Ġinterface . ĊĠĠĠĠĠĠĠ Ġsuper . handler ( this :: handle Subscription Message ) ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġhandler ( handler : ĠHandler < Response >): ĠRedis Connection Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠAvoid Ġuser Ġby passes Ġthe Ġimplementation ĊĠĠĠĠĠĠĠ Ġthrow ĠRedis He im d all Exception ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠReason . UNSUPPORTED _ ACTION , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" In Ġsubscription Ġmode , Ġyou Ġshould Ġnot Ġregister Ġyour Ġown Ġsubscription Ġhandler , Ġas Ġit 's Ġhandled Ġby ĠRedis ĠHe im d all ." ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġsend ( command : ĠRequest ): ĠFuture < Response > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpromise Ġ= ĠPromise . promise < Response >() ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( verify Subscription Command Only ( command . command (), Ġpromise )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . send ( command ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġpromise . future () Ġ// ĠFuture Ġalready Ġcompleted Ġhere Ġverify Subscription Commands Only Ġwill Ġcomplete Ġit ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġbatch ( commands : ĠList < Request >): ĠFuture < List < Response >> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpromise Ġ= ĠPromise . promise < List < Response >>() ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( verify Subscription Commands Only ( commands . map Ġ{ Ġit . command () Ġ}, Ġpromise )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . batch ( commands ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġpromise . future () Ġ// ĠFuture Ġalready Ġcompleted Ġhere Ġverify Subscription Commands Only Ġwill Ġcomplete Ġit ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsubscribe To Channels And Patterns ( ĊĠĠĠĠĠĠĠ Ġchannel Names : ĠList < String >, ĊĠĠĠĠĠĠĠ Ġchannel Patterns : ĠList < String >, ĊĠĠĠ Ġ): ĠFuture < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpromise Ġ= ĠPromise . promise < Unit >() ĊĠĠĠĠĠĠĠ Ġval Ġcommand F utures Ġ= ĠArrayList < Future <* >>() ĊĠĠĠĠĠĠĠ Ġif Ġ( channel Names . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= ĠRequest . cmd ( Command . SUB SCR IBE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchannel Names . addTo Command Args ( cmd ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= Ġsend ( cmd ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Channel ( s ) Ġ$ channel Names Ġsubscribed ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Failure Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Subscription Ġto Ġchannel ( s ) Ġ$ channel Names Ġfailed ", Ġit ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcommand F utures . add ( future ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( channel Patterns . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= ĠRequest . cmd ( Command . PS UB SCR IBE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchannel Patterns . addTo Command Args ( cmd ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= Ġsend ( cmd ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Channel Ġpattern ( s ) Ġ$ channel Patterns Ġsubscribed ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. on Failure Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Subscription Ġto Ġchannel Ġpattern ( s ) Ġ$ channel Patterns Ġfailed ", Ġit ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcommand F utures . add ( future ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( command F utures . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠComposite Future . all ( command F utures . toList ()). on Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpromise . complete () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. on Failure Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpromise . fail ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpromise . complete () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġpromise . future () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsubscribe After Re connect (): ĠFuture < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġchannel Names Ġ= Ġsubscription Store . subscriptions ()?. toList () Ġ?: Ġempty List () ĊĠĠĠĠĠĠĠ Ġval Ġpatterns Ġ= Ġsubscription Store . patterns ()?. toList () Ġ?: Ġempty List () ĊĠĠĠĠĠĠĠ Ġreturn Ġsubscribe To Channels And Patterns ( channel Names , Ġpatterns ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Subscription Message ( response : ĠResponse ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ response [ 0 ]}" Ġ== Ġ" subscribe " Ġ-> Ġsubscription Store . add Subscription ("${ response [ 1 ]} ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ response [ 0 ]}" Ġ== Ġ" unsubscribe " Ġ-> Ġsubscription Store . remove Subscription ("${ response [ 1 ]} ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ response [ 0 ]}" Ġ== Ġ" ps ub scribe " Ġ-> Ġsubscription Store . add Pattern ("${ response [ 1 ]} ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ response [ 0 ]}" Ġ== Ġ" p unsubscribe " Ġ-> Ġsubscription Store . remove Pattern ("${ response [ 1 ]} ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ response [ 0 ]}" Ġ== Ġ" message " Ġ-> Ġmessage Handler . handle ( response ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ response [ 0 ]}" Ġ== Ġ" p message " Ġ-> Ġmessage Handler . handle ( response ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġlogger . warn (" Response Ġ/ ĠMessage Ġtype Ġfrom ĠRedis Ġnot Ġsupported Ġ\"$ response \" ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ fun ĠList < String >. addTo Command Args ( command : ĠRequest ) Ġ= Ġcommand . apply Ġ{ ĠforEach Ġ{ Ġcommand . arg ( it ) Ġ} Ġ} Ċ