package Ġkv store . client Ċ Ċ import Ġra ft . service . Add Node Command Ċ import Ġra ft . service . Channel Ċ import Ġra ft . service . Channel Exception Ċ import Ġjava . io .* Ċ import Ġjava . net . Inet SocketAddress Ċ import Ġjava . net . Socket ĊĊ Ċ class ĠSocket Channel ( private Ġval Ġhost : ĠString , Ġprivate Ġval Ġport : ĠInt ) Ġ: ĠChannel Ġ{ ĊĠĠĠ Ġoverride Ġfun Ġsend ( payload : ĠAny ?): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSocket (). use Ġ{ Ġsocket Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket . tcp No Delay Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsocket . connect ( Inet SocketAddress ( host , Ġport )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( payload Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . write ( socket . get OutputStream (), Ġpayload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġread ( socket . getInputStream ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠChannel Exception (" failed Ġto Ġsend Ġand Ġreceive ", Ġe ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( IOException :: class ) ĊĠĠĠ Ġprivate Ġfun Ġread ( input : ĠInputStream ): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdata Input Ġ= ĠData InputStream ( input ) ĊĠĠĠĠĠĠĠ Ġval Ġmessage Type Ġ= Ġdata Input . readInt () ĊĠĠĠĠĠĠĠ Ġval Ġpayload Length Ġ= Ġdata Input . readInt () ĊĠĠĠĠĠĠĠ Ġval Ġpayload Ġ= ĠByteArray ( payload Length ) ĊĠĠĠĠĠĠĠ Ġdata Input . read Fully ( payload ) ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( message Type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Constants . MSG _ TYPE _ SUCCESS Ġ-> Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Constants . MSG _ TYPE _ FAILURE Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproto Failure : ĠProto s . Failure Ġ= ĠProto s . Failure . parseFrom ( payload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠChannel Exception ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" error Ġcode Ġ" Ġ+ Ġproto Failure . get ErrorCode (). toString () Ġ+ Ġ", Ġmessage Ġ" Ġ+ Ġproto Failure . getMessage () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Constants . MSG _ TYPE _ RE DIRECT Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproto Redirect : ĠProto s . Redirect Ġ= ĠProto s . Redirect . parseFrom ( payload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRedirect Exception ( NodeId ( proto Redirect . get Leader Id ())) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Constants . MSG _ TYPE _ GET _ COMMAND _ RESPONSE Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproto Get Command Response : ĠGet Command Response Ġ= ĠProto s . Get Command Response . parseFrom ( payload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! proto Get Command Response . get Found ()) Ġnull Ġelse Ġproto Get Command Response . getValue (). toByteArray () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠChannel Exception (" unexpected Ġmessage Ġtype Ġ$ message Type ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( IOException :: class ) ĊĠĠĠ Ġprivate Ġfun Ġwrite ( output : ĠOutputStream , Ġpayload : ĠAny ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( payload Ġis ĠGet Command ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproto Get Command : ĠGet Command Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠProto s . Get Command . newBuilder (). set Key (( payload Ġas ĠGet Command ). getKey ()). build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . write ( output , ĠMessage Constants . MSG _ TYPE _ GET _ COMMAND , Ġproto Get Command ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( payload Ġis ĠSet Command ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġset Command : ĠSet Command Ġ= Ġpayload Ġas ĠSet Command ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproto Set Command : ĠSet Command Ġ= ĠProto s . Set Command . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Key ( set Command . getKey ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setValue ( ByteString . copyFrom ( set Command . getValue ())). build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . write ( output , ĠMessage Constants . MSG _ TYPE _ SET _ COMMAND , Ġproto Set Command ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( payload Ġis ĠAdd Node Command ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommand : ĠAdd Node Command Ġ= Ġpayload Ġas ĠAdd Node Command ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproto Add Server Command : ĠAdd Node Command Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠProto s . Add Node Command . newBuilder (). set NodeId ( command . getNode Id ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Host ( command . getHost ()). set Port ( command . getPort ()). build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . write ( output , ĠMessage Constants . MSG _ TYPE _ ADD _ SERVER _ COMMAND , Ġproto Add Server Command ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( payload Ġis ĠRemove Node Command ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommand : ĠRemove Node Command Ġ= Ġpayload Ġas ĠRemove Node Command ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproto Remove Server Command : ĠRemove Node Command Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠProto s . Remove Node Command . newBuilder (). set NodeId ( command . getNode Id (). getValue ()). build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . write ( output , ĠMessage Constants . MSG _ TYPE _ REMOVE _ SERVER _ COMMAND , Ġproto Remove Server Command ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( IOException :: class ) ĊĠĠĠ Ġprivate Ġfun Ġwrite ( output : ĠOutputStream , Ġmessage Type : ĠInt , Ġmessage : ĠMessage Lite ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdata Output Ġ= ĠData OutputStream ( output ) ĊĠĠĠĠĠĠĠ Ġval Ġmessage Bytes : ĠByteArray Ġ= Ġmessage . toByteArray () ĊĠĠĠĠĠĠĠ Ġdata Output . writeInt ( message Type ) ĊĠĠĠĠĠĠĠ Ġdata Output . writeInt ( message Bytes . size ) ĊĠĠĠĠĠĠĠ Ġdata Output . write ( message Bytes ) ĊĠĠĠĠĠĠĠ Ġdata Output . flush () ĊĠĠĠ Ġ} Ċ Ċ } ĊĊ