package Ġru . spl ite . replic ator . demo Ċ Ċ import Ġcom . google . common . base . Stop watch Ċ import Ġio . ktor . application .* Ċ import Ġio . ktor . features .* Ċ import Ġio . ktor . http .* Ċ import Ġio . ktor . request .* Ċ import Ġio . ktor . response .* Ċ import Ġio . ktor . routing .* Ċ import Ġio . ktor . server . engine .* Ċ import Ġio . ktor . server . netty .* Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . sync . Semaphore Ċ import Ġkotlinx . coroutines . sync . with Permit Ċ import Ġkotlinx . coroutines . with Context Ċ import Ġkotlinx . serialization . json . Json Ċ import Ġorg . slf 4 j . Logger Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġru . spl ite . replic ator . demo . key value . KeyValue Command Ċ import Ġru . spl ite . replic ator . demo . key value . KeyValue Reply Ċ import Ġru . spl ite . replic ator . metrics . Metrics Ċ import Ġru . spl ite . replic ator . metrics . Metrics . record Stop watch Ċ import Ġru . spl ite . replic ator . state machine . StateMachine Command Submit ter Ċ import Ġru . spl ite . replic ator . transport . Node Identifier Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġkotlin . coroutines . Coroutine Context ĊĊ Ċ class ĠKey Value Store Controller ( ĊĠĠĠ Ġprivate Ġval Ġport : ĠInt , ĊĠĠĠ Ġprivate Ġval Ġnode Identifier : ĠNode Identifier , ĊĠĠĠ Ġmax Concurrent Submit s : ĠInt , ĊĠĠĠ Ġprivate Ġval Ġstate Machine Command Submit ter : ĠState Machine Command Submit ter < ByteArray , ĠByteArray >, ĊĠĠĠ Ġprivate Ġval Ġcoroutine Context : ĠCor outine Context Ġ= ĠDispatch ers . Un conf ined Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġsubmit Semaphore Ġ= ĠSem aphore ( max Concurrent Submit s ) ĊĊĠĠĠ Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠ Ġembedded Server ( Net ty , Ġport Ġ= Ġport ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġinstall ( Status Pages ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexception < Throwable > Ġ{ Ġcause Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( HttpStatusCode . InternalServerError , Ġcause . toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . error (" Error Ġwhile Ġhandling Ġrequest ", Ġcause ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġinstall ( Call Logging ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġinstall ( Default Headers ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Node - Identifier ", Ġnode Identifier . identifier ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġrouting Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġroute ("/ kv ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget ("{ key }") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Ġ= Ġcall . parameters [" key "] Ġ?: Ġerror (" Key Ġmust Ġbe Ġspecified ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= ĠKey Value Command . GetValue ( key ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Received ĠGET Ġcommand Ġ$ command ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommand Reply Ġ= Ġsubmit Command ( command ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond Text ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠJson . encode ToString ( KeyValue Reply . serializer (), Ġcommand Reply ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠContentType . Application . Json ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpost ("{ key }") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Ġ= Ġcall . parameters [" key "] Ġ?: Ġerror (" Key Ġmust Ġbe Ġspecified ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġvalue Ġ= Ġcall . receive Text () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= ĠKey Value Command . Put Value ( key , Ġvalue ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Received ĠPUT Ġcommand Ġ$ command ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommand Reply Ġ= Ġsubmit Command ( command ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond Text ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠJson . encode ToString ( KeyValue Reply . serializer (), Ġcommand Reply ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠContentType . Application . Json ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget (" health ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond Text (" OK ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. start ( wait Ġ= Ġfalse ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġsubmit Command ( command : ĠKey Value Command ): ĠKey Value Reply Ġ= Ġwith Context ( coroutine Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstop watch Ġ= ĠStop watch . create Started () ĊĠĠĠĠĠĠĠ Ġval Ġresponse Result Ġ= Ġkotlin . run Catch ing Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubmit Semaphore . with Permit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Bytes Ġ= Ġstate Machine Command Submit ter . submit ( KeyValue Command . serialize ( command )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠKey Value Reply . de serializer ( result Bytes ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġstop watch . stop () ĊĊĠĠĠĠĠĠĠ Ġif Ġ( response Result . is Success ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" Submitted Ġcommand Ġ$ command Ġin Ġ${ stop watch . elapsed ( TimeUnit . MILLISECONDS )} Ġms ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics . registry . command Submit Latency . record Stop watch ( stop watch ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics . registry . command Submit Error Latency . record Stop watch ( stop watch ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġresponse Result . get OrThrow () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠLOGGER : ĠLogger Ġ= ĠLoggerFactory . getLogger ( java Class . enc lo sing Class ) ĊĠĠĠ Ġ} Ċ }