package Ġio . small ry e . mut iny . coroutines Ċ Ċ import Ġio . small ry e . mut iny . Multi Ċ import Ġio . small ry e . mut iny . subscription . Multi Emitter Ċ import Ġjava . time . Duration Ċ import Ġjava . time . Instant Ċ import Ġjava . util . concurrent . Executors Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ import Ġkotlin . system . measure Time Millis Ċ import Ġkotlin . test . Test Ċ import Ġkotlin . test . assert Fails Ċ import Ġkotlin . test . assert Fails With Ċ import Ġkotlin . test . assertTrue Ċ import Ġkotlin . test . fail Ċ import Ġkotlinx . coroutines . Cancellation Exception Ċ import Ġkotlinx . coroutines . Experimental Cor outines Api Ċ import Ġkotlinx . coroutines . Timeout Cancellation Exception Ċ import Ġkotlinx . coroutines . as Coroutine Dispatcher Ċ import Ġkotlinx . coroutines . async Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . flow . buffer Ċ import Ġkotlinx . coroutines . flow . first Ċ import Ġkotlinx . coroutines . flow . toList Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġkotlinx . coroutines . with Timeout Ċ import Ġorg . assertj . core . api . Assertions . assertThat Ċ Ċ @ Experimental Cor outines Api Ċ class ĠMulti As Flow Test Ġ{ ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġimmediate Ġitems `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitems Ġ= ĠarrayOf ( 1 , Ġ 2 , Ġ 3 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). items (* items ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Items Ġ= Ġmulti . as Flow (). toList () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( event Items ). contains Exactly (* items ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġimmediate Ġfailure `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). failure < Any >( Exception (" k ab oom ")) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen Ġ& ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġmulti . as Flow () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Fails With < Exception >(" k ab oom ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow . toList () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġdelayed Ġitem Ġemits `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġitems Ġ= ĠarrayOf ( 5 , Ġ 2 3 , Ġ 4 2 ) ĊĠĠĠĠĠĠĠ Ġval Ġdelay Millis Ġ= Ġ 5 0 L ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). emitter Ġ{ Ġem : ĠMulti Emitter < in ĠInt > Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġembedded Scope (). launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġitems . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( delay Millis ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . emit ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . complete () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠ Ġval Ġ( before Subscribe , Ġevent Items , Ġafter Completion ) Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTrip le ( Instant . now (), Ġmulti . as Flow (). toList (), ĠInstant . now ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠ ĠassertThat ( event Items ). contains Exactly (* items ) ĊĠĠĠĠĠĠĠ ĠassertTrue ( Duration . between ( before Subscribe , Ġafter Completion ) Ġ>= ĠDuration . of Millis ( items . size Ġ* Ġdelay Millis )) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġdelayed Ġfailure `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġdelay Millis Ġ= Ġ 1 0 0 L ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). emitter Ġ{ Ġem : ĠMulti Emitter < in ĠInt > Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġembedded Scope (). launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( delay Millis ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . fail ( Exception (" boom ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen Ġ& ĠThen ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġmulti . as Flow () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġduration Millis Ġ= Ġmeasure Time Millis Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġassert Fails With < Exception >(" boom ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow . toList () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( duration Millis ). is GreaterThan OrEqualTo ( delay Millis ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġimmediate Ġcompletion `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). empty < Any >() ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Items Ġ= Ġmulti . as Flow (). toList () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( event Items ). isEmpty () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġfailure Ġafter Ġemit ting Ġan Ġitem `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġemitter Ġ= ĠAtomic Reference < Multi Emitter < in ĠInt >>() ĊĠĠĠĠĠĠĠ Ġfun Ġemit () Ġ= Ġemitter . get (). emit ( 5 ) ĊĠĠĠĠĠĠĠ Ġfun Ġfail () Ġ= Ġemitter . get (). fail ( IllegalStateException (" boom ")) ĊĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). emitter Ġ{ Ġem : ĠMulti Emitter < in ĠInt > Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġemitter . set ( em ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġmulti . as Flow () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Items Ġ= Ġmutable ListOf < Int >() ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġassert Job Ġ= Ġasync Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġassert Fails Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow . collect Ġ{ Ġevent Items . add ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġembedded Scope (). launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġyield Ġthread Ġand Ġwait Ġshortly Ġfor Ġensuring Ġemitter Ġgot Ġset ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfail () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( assert Job . await ()). has Message (" boom "). isInstanceOf ( IllegalStateException :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( event Items ). contains Exactly ( 5 , Ġ 5 ) ĊĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġcoroutine Ġcancellation Ġbefore Ġfirst Ġitem `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). ticks () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. starting After ( Duration . of Millis ( 1 5 0 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. every ( Duration . of Millis ( 5 0 )) ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠ Ġvar Ġretrieve Error : ĠThrowable ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjob Ġ= Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . as Flow (). toList () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfail () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( th : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretrieve Error Ġ= Ġth ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 5 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġjob . cancel ( Cancellation Exception (" abort ")) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠThread . sleep ( 3 5 0 ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠ ĠassertThat ( retrieve Error ). isInstanceOf ( Cancellation Exception :: class . java ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġcoroutine Ġcancellation Ġduring Ġitem Ġdelivery `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). ticks (). every ( Duration . of Millis ( 1 )) ĊĠĠĠĠĠĠĠ Ġval Ġcounter Ġ= ĠAtomicInteger () ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠ Ġvar Ġretrieve Error : ĠThrowable ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġtest Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjob Ġ= Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . as Flow (). collect Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcounter . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfail () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( th : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretrieve Error Ġ= Ġth ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġjob . cancel ( Cancellation Exception (" abort ")) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠThread . sleep ( 1 0 0 ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠ ĠassertThat ( retrieve Error ). isInstanceOf ( Cancellation Exception :: class . java ) ĊĠĠĠĠĠĠĠ ĠassertThat ( counter . get ()). is GreaterThan ( 0 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġcoroutine Ġtimeout Ġwhen Ġnothing Ġhappens `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). nothing < Any >() ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen Ġ& ĠThen ĊĠĠĠĠĠĠĠ Ġassert Fails With < Timeout Cancellation Exception > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtest Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwith Timeout ( 1 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . as Flow (). toList () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġempty ĠMulti `() Ġ{ ĊĠĠĠĠĠĠĠ Ġtest Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). empty < Any >() ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitems Ġ= Ġmulti . as Flow (). toList () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( items ). isEmpty () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġnull Ġitem `() Ġ{ ĊĠĠĠĠĠĠĠ Ġtest Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). item Ġ{ Ġnull Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitems Ġ= Ġmulti . as Flow (). toList () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ/// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( items ). has Size ( 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġbuffered Ġflow `() Ġ{ ĊĠĠĠĠĠĠĠ Ġtest Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitems Ġ= Ġ( 1 .. 1 0 _ 0 0 0 ). toList () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). iterable ( items ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġmulti . as Flow () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Items Ġ= Ġflow . buffer ( 5 ). toList () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( event Items ). contains Exactly Elements Of (( 1 .. 1 0 _ 0 0 0 ). toList ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` verify Ġthat Ġbuffer Ġoverflow Ġwith Ġslow Ġconsumer Ġworks Ġwell `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). ticks (). every ( Duration . of Millis ( 1 )) ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠ Ġval Ġticks Ġ= ĠAtomicInteger () ĊĠĠĠĠĠĠĠ Ġassert Fails With < Timeout Cancellation Exception > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtest Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwith Timeout ( 3 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . as Flow (). buffer ( 5 ). collect Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġticks . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 5 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠ ĠassertThat ( ticks . get ()). is Between ( 4 , Ġ 6 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` verify Ġthat ĠFlow Ġabort ion Ġcancel s Ġthe Ġsubscription Ġon Ġa Ġtick ing ĠMulti `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). ticks (). every ( Duration . of Millis ( 1 )) ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠ Ġval Ġtick Ġ= Ġtest Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . as Flow (). first () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠ ĠassertThat ( tick ). isEqualTo ( 0 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` verify Ġthat ĠFlow Ġabort ion Ġcancel s Ġthe Ġsubscription Ġon Ġan Ġemit ting ĠMulti `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). emitter < Int > Ġ{ Ġem Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( 1 .. 1 0 0 ). forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! em . is Cancelled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠThread . sleep ( 5 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . emit ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠ Ġval Ġstart Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠ Ġval Ġtick Ġ= Ġtest Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . as Flow (). first () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠ ĠassertThat ( tick ). isEqualTo ( 1 ) ĊĠĠĠĠĠĠĠ ĠassertThat ( System . currentTimeMillis () Ġ- Ġstart ). is LessThan ( 1 5 0 0 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` verify Ġthat ĠMulti Ġemission Ġis Ġnot Ġblocked Ġby Ġthe Ġcaller `() Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= ĠMulti . createFrom (). items ( 3 , Ġ 2 , Ġ 1 ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠ Ġval Ġlist Ġ= Ġtest Blocking ( Executors . new Single Thread Executor (). as Coroutine Dispatcher ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . as Flow (). toList () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠ ĠassertThat ( list ). contains Exactly ( 3 , Ġ 2 , Ġ 1 ) ĊĠĠĠ Ġ} Ċ } Ċ