package Ġio . small ry e . mut iny . coroutines Ċ Ċ import Ġio . small ry e . mut iny . helpers . test . Assert Subscriber Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . flow . flow Ċ import Ġkotlinx . coroutines . flow . flow Of Ċ import Ġorg . assertj . core . api . Assertions . assertThat Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ import Ġkotlin . test . Test Ċ Ċ class ĠFlow As Multi Test Ġ{ ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġimmediate Ġitem `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking ( Dispatch ers . Default ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitem Ġ= ĠUUID . randomUUID () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġflow Of ( item ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsubscriber Ġ= ĠAssert Subscriber . create < UUID >( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= Ġflow . as Multi () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . subscribe (). with Subscriber ( subscriber ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcollected Items Ġ= Ġmulti . collect (). asList (). await (). in def initely () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( collected Items ). contains Exactly ( item ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . await Completion (). assert Items ( item ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġimmediate Ġitems `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking ( Dispatch ers . Default ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitems Ġ= ĠarrayOf ( 5 , Ġ 2 3 , Ġ 4 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġflow Of (* items ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsubscriber Ġ= ĠAssert Subscriber . create < Int >( 5 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmulti Ġ= Ġflow . as Multi () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmulti . subscribe (). with Subscriber ( subscriber ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcollected Items Ġ= Ġmulti . collect (). asList (). await (). in def initely () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( collected Items ). contains Exactly (* items ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . await Completion (). assert Items (* items ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġimmediate Ġfailure `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking ( Dispatch ers . Default ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġflow < Any > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror (" boom ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsubscriber Ġ= ĠAssert Subscriber . create < Any >( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow . as Multi (). subscribe (). with Subscriber ( subscriber ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . await Failure () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. assert Failed With ( IllegalStateException :: class . java , Ġ" boom ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` verify Ġthat Ġcoroutine Ġcancellation Ġresult Ġin Ġfailure `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking ( Dispatch ers . Default ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġflow < UUID > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 2 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit ( UUID . randomUUID ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsubscriber Ġ= ĠAssert Subscriber . create < UUID >( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjob Ġ= Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow . as Multi (). subscribe (). with Subscriber ( subscriber ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 5 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjob . cancel ( Cancellation Exception (" abort ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠThread . sleep ( 3 5 0 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . await Failure (). assert Failed With ( Cancellation Exception :: class . java , Ġ" abort ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` verify Ġthat ĠFlow Ġcancel s Ġon ĠMulti Ġsubscription Ġcancellation `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking ( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠGiven ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcounter Ġ= ĠAtomicInteger () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġexit Exception : ĠThrowable ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġflow Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit ( counter . increment AndGet ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( err : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexit Exception Ġ Ġ= Ġerr ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsubscriber Ġ= ĠAssert Subscriber . create < Int >( 4 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow . as Multi (). subscribe (). with Subscriber ( subscriber ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 5 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 5 0 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThen ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( counter . get ()). is GreaterThan ( 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . assert Items (*( 1 .. 4 2 ). toList (). to Typed Array ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . assertNot Terminated () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( subscriber . is Cancelled ). isTrue () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( exit Exception ). isNotNull (). isInstanceOf ( Cancellation Exception :: class . java ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ