package Ġandroid . support . core . flow Ċ Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . Flow Collector Ċ import Ġkotlin . coroutines . Coroutine Context Ċ Ċ fun Ġ< T > Ġpipe Flow Of (): ĠPipe Flow < T > Ġ= ĠPipe Flow Impl () Ċ Ċ interface ĠPipe Flow < T > Ġ: ĠFlow < T > Ġ{ ĊĠĠĠ Ġfun Ġconnect ( flow : ĠFlow < T >) Ċ } Ċ Ċ suspend Ġfun Ġ< T > ĠFlow < T >. transport By ( flow : ĠPipe Flow < T >) Ġ{ ĊĠĠĠ Ġflow . connect ( this ) Ċ } Ċ Ċ private Ġclass ĠPipe Flow Impl < T > Ġ: ĠPipe Flow < T > Ġ{ ĊĠĠĠ Ġprivate Ġvar Ġm Source : ĠSource < T >? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġval Ġchannels Ġ= ĠCollector Channels < T >() ĊĊĠĠĠ Ġ@ Opt In ( Internal Cor outines Api :: class ) ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcollect ( collector : ĠFlow Collector < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= Ġchannels . alloc Slot ( collector ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcurrent Context Ġ= Ġcurrent Coroutine Context ()[ Job ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcurrent Context !!. ensure Active () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsource Ġ= Ġm Source ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠDouble Ġcheck Ġto Ġreduce Ġlock Ġtime ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( source Ġ!= Ġnull Ġ&& Ġ! source . is Collected ( collector )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( this @ Pipe Flow Impl ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! source . is Collected ( collector )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsource . collect ( collector , Ġcurrent Context ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . await () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchannels . free Slot ( collector ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġconnect ( flow : ĠFlow < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġm Source ?. cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġm Source Ġ= ĠSource ( flow ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchannels . try Notify () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠSource < T >( val Ġflow : ĠFlow < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġm Collected Ġ= Ġhash Map Of < Flow Collector < T >, ĠBoolean >() ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġm Job : ĠJob ? Ġ= Ġnull ĊĊĠĠĠĠĠĠĠ Ġ@ Internal Cor outines Api ĊĠĠĠĠĠĠĠ Ġfun Ġcollect ( collector : ĠFlow Collector < T >, Ġcontext : ĠCor outine Context ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( m Collected ) Ġ{ Ġm Collected [ collector ] Ġ= Ġtrue Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġm Job Ġ= ĠCor outine Scope ( context ). launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġflow . collect ( collector ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġcancel () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġm Job ?. cancel () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġis Collected ( collector : ĠFlow Collector < T >): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġsynchronized ( m Collected ) Ġ{ Ġm Collected [ collector ] Ġ?: Ġfalse Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ }