package Ġcom . jetbrains . rd . rd text . test . util Ċ Ċ import Ġcom . jetbrains . rd . framework . Abstract Buffer Ċ import Ġcom . jetbrains . rd . framework . Rd Id Ċ import Ġcom . jetbrains . rd . framework . base . Wire Base Ċ import Ġcom . jetbrains . rd . framework . create Abstract Buffer Ċ import Ġcom . jetbrains . rd . util . Queue Ċ import Ġcom . jetbrains . rd . util . reactive . IS cheduler ĊĊ Ċ class ĠTest Wire ( scheduler Ġ: ĠIS cheduler ) Ġ: ĠWire Base ( scheduler ) Ġ{ ĊĠĠĠ Ġlateinit Ġvar Ġcounter part Ġ: ĠTest Wire ĊĊĠĠĠ Ġval Ġmsg Q Ġ= ĠQueue < Rd Message >() ĊĊĠĠĠ Ġvar Ġbytes Written : ĠLong Ġ= Ġ 0 ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġconnected . set ( true ) ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġoverride Ġfun Ġsend ( id : ĠR d Id , Ġwriter : Ġ( Abstract Buffer ) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire (! id . isNull ) ĊĊĠĠĠĠĠĠĠ Ġval Ġo stream Ġ= Ġcreate Abstract Buffer () ĊĠĠĠĠĠĠĠ Ġcontexts . write Current Message Context ( ostream ) ĊĠĠĠĠĠĠĠ Ġwriter ( ostream ) ĊĊĠĠĠĠĠĠĠ Ġbytes Written Ġ+= Ġo stream . position ĊĊĠĠĠĠĠĠĠ Ġo stream . position Ġ= Ġ 0 ĊĊĠĠĠĠĠĠĠ Ġmsg Q . offer ( Rd Message ( id , Ġo stream )) ĊĠĠĠĠĠĠĠ Ġif Ġ( auto Flush ) Ġprocess All Messages () ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġvar Ġauto Flush Ġ: ĠBoolean Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġfield ĊĠĠĠĠĠĠĠ Ġset ( value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfield Ġ= Ġvalue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( value ) Ġprocess All Messages () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġprocess All Messages () Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ(! msg Q . isEmpty ()) Ġprocess One Message () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġprocess One Message () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= Ġmsg Q . poll () Ġ?: Ġreturn ĊĠĠĠĠĠĠĠ Ġcounter part . message Broker . dispatch ( msg . id , Ġmsg . istream ) ĊĠĠĠ Ġ} Ċ } ĊĊ Ċ class ĠR d Message Ġ( val Ġid Ġ: ĠR d Id , Ġval Ġist ream Ġ: ĠAbstract Buffer ) Ċ