package Ġcom . hoc . flow m vi . core Ċ Ċ import Ġkotlinx . coroutines . Cancellation Exception Ċ import Ġkotlinx . coroutines . Experimental Cor outines Api Ċ import Ġkotlinx . coroutines . cancel Ċ import Ġkotlinx . coroutines . coroutine Scope Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . as Flow Ċ import Ġkotlinx . coroutines . flow . catch Ċ import Ġkotlinx . coroutines . flow . channel Flow Ċ import Ġkotlinx . coroutines . flow . collect Ċ import Ġkotlinx . coroutines . flow . flow Ċ import Ġkotlinx . coroutines . flow . map Ċ import Ġkotlinx . coroutines . flow . on Each Ċ import Ġkotlinx . coroutines . flow . take Ċ import Ġkotlinx . coroutines . launch Ċ import Ġjava . util . concurrent . atomic . Atomic Boolean Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ Ċ @ Experimental Cor outines Api Ċ fun Ġ< T , ĠR > ĠFlow < T >. take Until ( notifier : ĠFlow < R >): ĠFlow < T > Ġ= Ġchannel Flow Ġ{ ĊĠ Ġval Ġouter Scope Ġ= Ġthis ĊĊĠ Ġlaunch Ġ{ ĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġnot ifier . take ( 1 ). collect () ĊĠĠĠĠĠ Ġclose () ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠ Ġouter Scope . cancel ( e ) Ġ// Ġcancel Ġouter Ġscope Ġon Ġcancellation Ġexception , Ġtoo ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĠ Ġlaunch Ġ{ ĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġcollect Ġ{ Ġsend ( it ) Ġ} ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠ Ġouter Scope . cancel ( e ) Ġ// Ġcancel Ġouter Ġscope Ġon Ġcancellation Ġexception , Ġtoo ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ @ Experimental Cor outines Api Ċ fun Ġ< T , ĠR > ĠFlow < T >. flatMap First ( transform : Ġsuspend Ġ( value : ĠT ) Ġ-> ĠFlow < R >): ĠFlow < R > Ġ= ĊĠ Ġmap ( transform ). flatten First () Ċ Ċ @ Experimental Cor outines Api Ċ fun Ġ< T > ĠFlow < Flow < T >> . flatten First (): ĠFlow < T > Ġ= Ġchannel Flow Ġ{ ĊĠ Ġval Ġouter Scope Ġ= Ġthis ĊĠ Ġval Ġbusy Ġ= ĠAtomic Boolean ( false ) ĊĠ Ġcollect Ġ{ Ġinner Ġ-> ĊĠĠĠ Ġif Ġ( busy . compare AndSet ( false , Ġtrue )) Ġ{ ĊĠĠĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġinner . collect Ġ{ Ġouter Scope . send ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġbusy . set ( false ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġ// Ġcancel Ġouter Ġscope Ġon Ġcancellation Ġexception , Ġtoo ĊĠĠĠĠĠĠĠĠĠ Ġouter Scope . cancel ( e ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ private Ġobject ĠUN INITIAL IZED Ċ Ċ fun Ġ< A , ĠB , ĠR > ĠFlow < A >. with Latest From ( other : ĠFlow < B >, Ġtransform : Ġsuspend Ġ( A , ĠB ) Ġ-> ĠR ): ĠFlow < R > Ġ{ ĊĠ Ġreturn Ġflow Ġ{ ĊĠĠĠ Ġcoroutine Scope Ġ{ ĊĠĠĠĠĠ Ġval Ġlatest B Ġ= ĠAtomic Reference < Any >( UN INITIAL IZED ) ĊĠĠĠĠĠ Ġval Ġouter Scope Ġ= Ġthis ĊĊĠĠĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġother . collect Ġ{ Ġlatest B . set ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġouter Scope . cancel ( e ) Ġ// Ġcancel Ġouter Ġscope Ġon Ġcancellation Ġexception , Ġtoo ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġcollect Ġ{ Ġa Ġ-> ĊĠĠĠĠĠĠĠ Ġval Ġb Ġ= Ġlatest B . get () ĊĠĠĠĠĠĠĠ Ġif Ġ( b Ġ!= ĠUN INITIAL IZED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠ Ġemit ( transform ( a , Ġb Ġas ĠB )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ @ Experimental Cor outines Api Ċ suspend Ġfun Ġmain () Ġ{ ĊĠ Ġ( 1 .. 1 0 0 ). as Flow () ĊĠĠĠ Ġ. on Each Ġ{ Ġdelay ( 5 0 ) Ġ} ĊĠĠĠ Ġ. take Until ( flow Ġ{ Ġdelay ( 2 0 0 ); Ġemit ( Unit ) Ġ}) ĊĠĠĠ Ġ. collect Ġ{ Ġprintln ("> >>>> Ġ$ it ") Ġ} ĊĊĠ Ġprintln (" Done ") ĊĊĠ Ġ( 1 .. 2 0 0 0 ). as Flow () ĊĠĠĠ Ġ. on Each Ġ{ Ġdelay ( 5 0 ) Ġ} ĊĠĠĠ Ġ. flatMap First Ġ{ Ġv Ġ-> ĊĠĠĠĠĠ Ġflow Ġ{ ĊĠĠĠĠĠĠĠ Ġdelay ( 5 0 0 ) ĊĠĠĠĠĠĠĠ Ġemit ( v ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠ Ġ. on Each Ġ{ Ġprintln ("[ * ] Ġ$ it ") Ġ} ĊĠĠĠ Ġ. catch Ġ{ Ġprintln (" Error Ġ$ it ") Ġ} ĊĠĠĠ Ġ. collect () Ċ } Ċ