package Ġcom . example . rx kotlin _ network Ċ Ċ import Ġio . reactivex .* Ċ import Ġjava . io . BufferedReader Ċ import Ġjava . io . InputStream Ċ import Ġjava . io . InputStream Reader Ċ import Ġjava . lang . Thread Ċ import Ġjava . util . Optional Ċ import Ġjava . util . concurrent . Linked BlockingQueue Ċ Ċ class ĠNetwork Read Thread Ġ: ĠThread Ġ{ ĊĠĠĠ Ġval ĠoutputStream : ĠFlow able < String > ĊĊĠĠĠ Ġprivate Ġval ĠinputStream : ĠBufferedReader ĊĠĠĠ Ġprivate Ġval Ġqueue Ġ= ĠLinked BlockingQueue < Optional < String >>() ĊĊĠĠĠ Ġconstructor ( inputStream : ĠInputStream ) Ġ: Ġsuper () Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠTODO ĠBuffered InputStream Ġfor Ġbyte Ġdata ? ĊĠĠĠĠĠĠĠ Ġ// Ġhttps :// stackoverflow . com / questions / 1 5 5 3 8 5 0 9 / de aling - with - end - of - file - using - buffered reader - read ĊĠĠĠĠĠĠĠ Ġthis . inputStream Ġ= ĠBufferedReader ( InputStream Reader ( inputStream )) ĊĠĠĠĠĠĠĠ Ġthis . output Stream Ġ= ĠFlow able . create ({ Ġsubscriber Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO ĠWait Ġuntil Ġthread Ġis Ġalive Ġusing ĠCount DownLatch ? ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO Ġthread . is Alive ()? Ġsocket . isOpen ? Ġvolatile Ġboolean Ġflag ? ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitem Ġ= Ġqueue . take () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! item . isPresent ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . onNext ( item . get ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . onComplete () ĊĠĠĠĠĠĠĠĠ Ġ}, ĠBack pressure Strategy . MISSING ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdata Ġ= ĠinputStream . readLine () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( data Ġ== Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Found Ġdata ! Ġ$ data ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . put ( Optional . of ( data )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠInterruptedException ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . put ( Optional . empty ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ