package Ġcom . app . missed notifications reminder . util . flow Ċ Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . channels . Closed Receive Channel Exception Ċ import Ġkotlinx . coroutines . channels . Receive Channel Ċ import Ġkotlinx . coroutines . channels . produce Ċ import Ġkotlinx . coroutines . channels . ticker Ċ import Ġkotlinx . coroutines . flow . Abstract Flow Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . Flow Collector Ċ import Ġkotlinx . coroutines . flow . collect Ċ import Ġkotlinx . coroutines . select s . select Ċ Ċ /** Ċ Ġ* ĠImplementation Ġis Ġtaken Ġfrom Ċ Ġ* Ġhttps :// dev . to / ps f eng / a - story - of - building - a - custom - flow - operator - buff ert ime out - 4 d 9 5 Ċ Ġ*/ Ċ @ Flow Preview Ċ internal Ġclass ĠFlow Buffer Timeout < T > Ġconstructor ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsource : ĠFlow < T >, ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsize : ĠInt , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġduration : ĠLong ) Ġ: ĠAbstract Flow < List < T >>() Ġ{ ĊĠĠĠ Ġ@ Experimental Cor outines Api ĊĠĠĠ Ġ@ Obsolete Cor outines Api ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcollect S af ely ( collector : ĠFlow Collector < List < T >>) Ġ{ ĊĠĠĠĠĠĠĠ Ġcoroutine Scope Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevents Ġ= Ġmutable ListOf < T >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġticker Channel Ġ= Ġticker ( duration ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġupstream Values : ĠReceive Channel < T > Ġ= Ġproduce Ġ{ Ġsource . collect Ġ{ Ġvalue Ġ-> Ġsend ( value ) Ġ} Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( isActive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġhas Timed Out Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġselect < Unit > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupstream Values . on Receive Ġ{ Ġvalue Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents . add ( value ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġticker Channel . on Receive Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhas Timed Out Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( events . size Ġ== Ġsize Ġ|| Ġ( has Timed Out Ġ&& Ġevents . isNotEmpty ())) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcollector . emit ( events ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevents . clear () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠClosed Receive Channel Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġdrain Ġremaining Ġevents ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( events . isNotEmpty ()) Ġcollector . emit ( events ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġticker Channel . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }