package Ġcom . ja ffa . rpc . lib . rabbitmq Ċ Ċ import Ġcom . ja ffa . rpc . lib . common . Option Constants Ċ import Ġcom . ja ffa . rpc . lib . entities . Protocol Ċ import Ġcom . ja ffa . rpc . lib . exception . J aff a Rpc ExecutionException Ċ import Ġcom . ja ffa . rpc . lib . exception . J aff a Rpc No Route Exception Ċ import Ġcom . ja ffa . rpc . lib . exception . J aff a Rpc System Exception Ċ import Ġcom . ja ffa . rpc . lib . request . Sender Ċ import Ġcom . ja ffa . rpc . lib . zookeeper . Utils Ċ import Ġcom . rabbitmq . client . AM QP Ċ import Ġcom . rabbitmq . client . Channel Ċ import Ġcom . rabbitmq . client . Consumer Ċ import Ġcom . rabbitmq . client . Default Consumer Ċ import Ġcom . rabbitmq . client . Envelope Ċ import Ġorg . apache . commons . lang 3 . StringUtils Ċ import Ġorg . springframework . amqp . A mq p Exception Ċ import Ġorg . springframework . amqp . rabbit . connection . Connection Ċ import Ġjava . io . IOException Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ Ċ class ĠRabbit MQ Request Sender Ġ: ĠSender () Ġ{ ĊĊĠĠĠ Ġpublic Ġoverride Ġfun Ġexecute Sync ( message : ĠByteArray ?): ĠByteArray ? Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġatomic Reference Ġ= ĠAtomic Reference < ByteArray >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequests [ command . rq Uid ] Ġ= Ġobject Ġ: ĠCallback Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġcall ( body : ĠByteArray ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġatomic Reference . set ( body ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Sync ( message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstart Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ(!( timeout Ġ!= Ġ- 1 L Ġ&& ĠSystem . currentTimeMillis () Ġ- Ġstart Ġ> Ġtimeout Ġ|| ĠSystem . currentTimeMillis () Ġ- Ġstart Ġ> Ġ 1 0 0 0 Ġ* Ġ 6 0 Ġ* Ġ 6 0 )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġatomic Reference . get () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( result Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequests . remove ( command . rq Uid ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequests . remove ( command . rq Uid ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ja ffa Rpc No Route Exception : ĠJ aff a Rpc No Route Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġj aff a Rpc No Route Exception ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( exception : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠJ aff a Rpc ExecutionException ( exception ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġnull ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( IOException :: class ) ĊĠĠĠ Ġprivate Ġfun Ġsend Sync ( message : ĠByteArray ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtarget Module Id : ĠString ? ĊĠĠĠĠĠĠĠ Ġif Ġ( StringUtils . isNotBlank ( module Id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtarget Module Id Ġ= Ġmodule Id ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUtils . getHost For Service ( Utils . getService Interface Name From Client ( command . service Class ), Ġmodule Id , ĠProtocol . R AB BIT ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtarget Module Id Ġ= ĠUtils . getModule For Service ( Utils . getService Interface Name From Client ( command . service Class ), ĠProtocol . R AB BIT ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġclient Channel . basic Publish ( target Module Id , Ġ"$ target Module Id - server ", Ġnull , Ġmessage ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġpublic Ġoverride Ġfun Ġexecute Async ( message : ĠByteArray ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsend Sync ( message ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ja ffa Rpc No Route Exception : ĠJ aff a Rpc No Route Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġj aff a Rpc No Route Exception ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠJ aff a Rpc ExecutionException ( e ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġinterface ĠCallback Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġcall ( body : ĠByteArray ?) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠNAME _ PREFIX Ġ= ĠOption Constants . MODULE _ ID ĊĠĠĠĠĠĠĠ Ġval ĠEX CHANGE _ NAME Ġ= ĠNAME _ PREFIX ĊĠĠĠĠĠĠĠ Ġval ĠCLIENT _ SYNC _ NAME Ġ= Ġ"$ NAME _ PREFIX - client - sync " ĊĠĠĠĠĠĠĠ Ġval ĠCLIENT _ ASYNC _ NAME Ġ= Ġ"$ NAME _ PREFIX - client - async " ĊĠĠĠĠĠĠĠ Ġval ĠSERVER Ġ= Ġ"$ NAME _ PREFIX - server " ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġrequests : ĠMutable Map < String ?, ĠCallback > Ġ= ĠConcurrent HashMap () ĊĠĠĠĠĠĠĠ Ġprivate Ġlateinit Ġvar Ġconnection : ĠConnection ĊĠĠĠĠĠĠĠ Ġprivate Ġlateinit Ġvar Ġclient Channel : ĠChannel ĊĊĠĠĠĠĠĠĠ Ġ@ kotlin . jvm . Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġinit () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection Ġ= ĠRabbit MQ Protocol Controller . connection Factory . create Connection () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Channel Ġ= Ġconnection . create Channel ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Channel . queue Bind ( CLIENT _ SYNC _ NAME , ĠEX CHANGE _ NAME , ĠCLIENT _ SYNC _ NAME ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconsumer : ĠConsumer Ġ= Ġobject Ġ: ĠDefault Consumer ( client Channel ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Throws ( IOException :: class ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġhandle Delivery ( consumer Tag : ĠString , Ġenvelope : ĠEn velope , Ġproperties : ĠAMQP . Basic Properties , Ġbody : ĠByteArray ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( properties . correlation Id Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcallback Ġ= Ġrequests . remove ( properties . correlation Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( callback Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcallback . call ( body ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Channel . basic Ack ( envelope . delivery Tag , Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Channel . basic Consume ( CLIENT _ SYNC _ NAME , Ġfalse , Ġconsumer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( io Exception : ĠA mq p Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠJ aff a Rpc System Exception ( io Exception ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( io Exception : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠJ aff a Rpc System Exception ( io Exception ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Channel . close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ignore : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNo - op ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }