package Ġorg . jav ay nc . io Ċ Ċ import Ġkotlinx . coroutines . cancel Ċ import Ġkotlinx . coroutines . channels . Producer Scope Ċ import Ġkotlinx . coroutines . channels . await Close Ċ import Ġkotlinx . coroutines . channels . send Blocking Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . buffer Ċ import Ġkotlinx . coroutines . flow . callback Flow Ċ import Ġorg . reactiv est ream s . Subscriber Ċ import Ġorg . reactiv est ream s . Subscription Ċ import Ġjava . nio . channels . As ynchronous File Channel Ċ import Ġjava . nio . file . Path Ċ import Ġjava . nio . file . Standard Open Option Ċ Ċ class ĠAsync FileReader Flow ( private Ġval Ġchannel : ĠProducer Scope < String >) Ġ: ĠSubscriber < String > Ġ{ ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġsign : ĠSubscription ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Jvm Static Ġfun Ġlines ( file : ĠPath ) Ġ: ĠFlow < String > Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcallback Flow < String > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsub Ġ= ĠAsync FileReader Flow ( this ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAsync Files . lines ( file ). subscribe ( sub ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠNext Ġawait Close () Ġis Ġequivalent Ġto Ġcontinuation . invoke On Cancellation Ġans Ġit Ġis Ġmandatory . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ* ĠShould Ġbe Ġused Ġin Ġthe Ġend Ġof Ġcallback Flow Ġblock . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġawait Close Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsub . sign . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. buffer ( Abstract Async FileReader Lines . BUFFER _ SIZE ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonError ( error : ĠThrowable ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġchannel . close ( error ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ ĊĠĠĠĠĠĠĠ Ġchannel . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Subscribe ( subscription : ĠSubscription ) Ġ{ ĊĠĠĠĠĠĠĠ Ġsign Ġ= Ġsubscription ĊĠĠĠĠĠĠĠ Ġsign . request ( Long . MAX _ VALUE ); ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Next ( line : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠRe pl acing Ġthe Ġfollowing Ġsend Blocking () Ġwith Ġoffer () Ġ( which Ġis Ġasync ) ĊĠĠĠĠĠĠĠĠ Ġ* Ġwill Ġrequire Ġthe Ġuse Ġof Ġa Ġchained Ġbuffer () Ġin Ġresulting ĠFlow Ġpipeline ĊĠĠĠĠĠĠĠĠ Ġ* Ġto Ġavoid Ġlo osing Ġemitted Ġvalues . ĊĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġsend Blocking () Ġapproach Ġis Ġthrowing ĠAbort Flow Exception , Ġapparently ĊĠĠĠĠĠĠĠĠ Ġ* Ġmaybe Ġbecause Ġsomething Ġhas Ġbeen Ġemitted Ġafter Ġcancellation Ġrequest . ĊĠĠĠĠĠĠĠĠ Ġ* ĠIt Ġhappens Ġwith Ġtake While (). ĠMaybe Ġwe Ġcan Ġreproduce Ġit Ġa Ġin Ġsmall Ġscenario . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġchannel . offer ( line ) ĊĠĠĠĠĠĠĠ Ġ// Ġchannel . send Blocking ( line ) ĊĠĠĠ Ġ} Ċ }