// ĠCopyright Ġ 2 0 2 0 ĠThe ĠCross - Media ĠMeasurement ĠAuthors Ċ // Ċ // ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ // Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ // ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ // Ċ // ĠĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ // Ċ // ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ // Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ // ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ // ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ // Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ċ package Ġorg . wf an et . measurement . common Ċ Ċ import Ġcom . google . common . truth . Truth . assertThat Ċ import Ġkotlin . test . assert Fails With Ċ import Ġkotlin . test . assertFalse Ċ import Ġkotlin . test . assertNotNull Ċ import Ġkotlin . test . assertTrue Ċ import Ġkotlin . test . fail Ċ import Ġkotlinx . coroutines . flow . as Flow Ċ import Ġkotlinx . coroutines . flow . flow Of Ċ import Ġkotlinx . coroutines . flow . toList Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġorg . junit . Test Ċ import Ġorg . junit . runner . RunWith Ċ import Ġorg . junit . runners . JUnit 4 Ċ Ċ @ RunWith ( JUnit 4 :: class ) Ċ class ĠFlow Kt Test Ġ{ ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` with Retries On Each Ġno Ġerrors Ġthrown `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġsuccessfully Processed Items Ġ= Ġmutable ListOf < Int >() ĊĠĠĠ Ġvar Ġtotal T ries Ġ= Ġ 0 ĊĠĠĠ Ġ( 1 .. 1 0 ) ĊĠĠĠĠĠ Ġ. as Flow () ĊĠĠĠĠĠ Ġ. with Retries On Each ( 3 , Ġretry Predicate Ġ= Ġ{ Ġfail (" unexpected Ġretry ") Ġ}) Ġ{ Ġtotal T ries Ġ+= Ġ 1 Ġ} ĊĠĠĠĠĠ Ġ. toList ( successfully Processed Items ) ĊĠĠĠ ĠassertThat ( successfully Processed Items ). isEqualTo (( 1 .. 1 0 ). toList ()) ĊĠĠĠ ĠassertThat ( total T ries ). isEqualTo ( 1 0 ) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` with Retries On Each Ġnon - ret ri able Ġerror `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġsuccessfully Processed Items Ġ= Ġmutable ListOf < Int >() ĊĠĠĠ Ġval Ġtries Counter Ġ= Ġmutable Map Of < Int , ĠInt >() ĊĠĠĠ Ġassert Fails With < IllegalArgumentException > Ġ{ ĊĠĠĠĠĠ Ġ( 1 .. 1 0 0 ) ĊĠĠĠĠĠĠĠ Ġ. as Flow () ĊĠĠĠĠĠĠĠ Ġ. with Retries On Each ( 3 , Ġretry Predicate Ġ= Ġ{ Ġit Ġis ĠNullPointerException Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġtries Counter . put If Absent ( it , Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠ Ġtries Counter [ it ] Ġ= Ġ 1 Ġ+ Ġ( tri es Counter [ it ] Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠ Ġrequire ( it Ġ< Ġ 5 ) Ġ{ Ġ" Sim ulated Ġerror Ġin Ġtest ..." Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠThe Ġrequire () Ġwill Ġthrow Ġan ĠIllegalArgumentException Ġthat Ġis Ġnot Ġcaught Ġin Ġthe Ġflow . ĊĠĠĠĠĠĠĠ Ġ// ĠIt Ġwill Ġbe Ġthrown Ġon Ġthe Ġcollection Ġof Ġthis Ġflow . ĊĠĠĠĠĠĠĠ Ġ. toList ( successfully Processed Items ) ĊĠĠĠ Ġ} ĊĠĠĠ ĠassertThat ( successfully Processed Items ). isEqualTo (( 1 .. 4 ). toList ()) ĊĠĠĠ ĠassertThat ( tri es Counter ). isEqualTo ( map Of ( 1 Ġto Ġ 1 , Ġ 2 Ġto Ġ 1 , Ġ 3 Ġto Ġ 1 , Ġ 4 Ġto Ġ 1 , Ġ 5 Ġto Ġ 1 )) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` with Retries On Each Ġretry Ġerrors `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġsuccessfully Processed Items Ġ= Ġmutable ListOf < Int >() ĊĠĠĠ Ġval Ġtries Counter Ġ= Ġmutable Map Of < Int , ĠInt >() ĊĠĠĠ Ġassert Fails With < IllegalArgumentException > Ġ{ ĊĠĠĠĠĠ Ġ( 1 .. 1 0 0 ) ĊĠĠĠĠĠĠĠ Ġ. as Flow () ĊĠĠĠĠĠĠĠ Ġ. with Retries On Each ( 3 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġtries Counter . put If Absent ( it , Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠ Ġtries Counter [ it ] Ġ= Ġ 1 Ġ+ Ġ( tri es Counter [ it ] Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠ Ġrequire ( it Ġ< Ġ 8 ) Ġ{ Ġ" Sim ulated Ġerror Ġin Ġtest ..." Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠThe Ġrequire () Ġwill Ġthrow Ġan ĠIllegalArgumentException Ġthat Ġis Ġcaught Ġin Ġthe Ġflow , Ġand ĊĠĠĠĠĠĠĠ Ġ// Ġmatches Ġthe Ġpredicate Ġcausing Ġthe Ġblock Ġto Ġbe Ġretried . ĠHowever , Ġit Ġwill Ġbe Ġtried Ġ 3 Ġtimes ĊĠĠĠĠĠĠĠ Ġ// Ġand Ġthe Ġexception Ġwill Ġbe Ġthrown Ġall Ġthree Ġtimes Ġat Ġwhich Ġpoint Ġit Ġis Ġre throw n . ĊĠĠĠĠĠĠĠ Ġ. toList ( successfully Processed Items ) ĊĠĠĠ Ġ} ĊĠĠĠ ĠassertThat ( successfully Processed Items ). isEqualTo (( 1 .. 7 ). toList ()) ĊĠĠĠ ĠassertThat ( tri es Counter ) ĊĠĠĠĠĠ Ġ. isEqualTo ( ĊĠĠĠĠĠĠĠ Ġ// Ġall Ġitems Ġshould Ġhave Ġbeen Ġattempted Ġonce Ġexcept Ġthe Ġlast Ġitem Ġwhich Ġis Ġattempted Ġ 3 ĊĠĠĠĠĠĠĠ Ġ// Ġtimes . ĊĠĠĠĠĠĠĠ Ġmap Of ( 1 Ġto Ġ 1 , Ġ 2 Ġto Ġ 1 , Ġ 3 Ġto Ġ 1 , Ġ 4 Ġto Ġ 1 , Ġ 5 Ġto Ġ 1 , Ġ 6 Ġto Ġ 1 , Ġ 7 Ġto Ġ 1 , Ġ 8 Ġto Ġ 3 ) ĊĠĠĠĠĠ Ġ) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` with Retries On Each Ġretries Ġand Ġsucceeds `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġsuccessfully Processed Items Ġ= Ġmutable ListOf < Int >() ĊĠĠĠ Ġval Ġtries Counter Ġ= Ġmutable Map Of < Int , ĠInt >() ĊĠĠĠ Ġ( 1 .. 6 ) ĊĠĠĠĠĠ Ġ. as Flow () ĊĠĠĠĠĠ Ġ. with Retries On Each ( 3 ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġinit Value Ġ= Ġtries Counter . put If Absent ( it , Ġ 0 ) ĊĠĠĠĠĠĠĠ Ġtries Counter [ it ] Ġ= Ġ 1 Ġ+ Ġ( tri es Counter [ it ] Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠ Ġrequire ( it Ġ% Ġ 2 Ġ== Ġ 0 Ġ|| Ġinit Value Ġ!= Ġnull ) Ġ{ Ġ" Sim ulated Ġerror Ġin Ġtest ..." Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ// ĠThe Ġrequire () Ġwill Ġthrow Ġan ĠIllegalArgumentException Ġthat Ġis Ġcaught Ġin Ġthe Ġflow , Ġand ĊĠĠĠĠĠ Ġ// Ġmatches Ġthe Ġpredicate Ġcausing Ġthe Ġblock Ġto Ġbe Ġretried . ĠHowever , Ġit Ġsucceeds Ġon Ġretry . ĊĠĠĠĠĠ Ġ. toList ( successfully Processed Items ) ĊĠĠĠ ĠassertThat ( successfully Processed Items ). isEqualTo (( 1 .. 6 ). toList ()) ĊĠĠĠ Ġ// ĠEven Ġitems Ġare Ġattempted Ġonce Ġand Ġodd Ġitems Ġattempted Ġtwice . ĊĠĠĠ ĠassertThat ( tri es Counter ). isEqualTo ( map Of ( 1 Ġto Ġ 2 , Ġ 2 Ġto Ġ 1 , Ġ 3 Ġto Ġ 2 , Ġ 4 Ġto Ġ 1 , Ġ 5 Ġto Ġ 2 , Ġ 6 Ġto Ġ 1 )) ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` concurrent Map Ġruns Ġconcurrent ly `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġl atch Ġ= ĠCount DownLatch ( 1 0 0 ) ĊĠĠĠ Ġval Ġevents Ġ= Ġmutable ListOf < Int >() ĊĠĠĠ Ġval Ġresult Ġ= ĊĠĠĠĠĠ Ġ( 1 .. 1 0 0 ) ĊĠĠĠĠĠĠĠ Ġ. as Flow () ĊĠĠĠĠĠĠĠ Ġ. map Concurrent ly ( this , Ġ 1 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġevents . add ( it ) ĊĠĠĠĠĠĠĠĠĠ Ġl atch . count Down () ĊĠĠĠĠĠĠĠĠĠ Ġl atch . await () ĊĠĠĠĠĠĠĠĠĠ Ġ- it ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ. toList () ĊĠĠĠ ĠassertThat ( events ). contains Exactly Elements In ( 1 .. 1 0 0 ). in Order () ĊĠĠĠ ĠassertThat ( result ). contains Exactly Elements In (- 1 Ġdown To Ġ- 1 0 0 ). in Order () ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` concurrent Map Ġruns Ġin Ġorder `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġevents Ġ= Ġmutable ListOf < Int >() ĊĠĠĠ Ġval Ġresult Ġ= ĊĠĠĠĠĠ Ġflow Of ( 1 , Ġ 2 ) ĊĠĠĠĠĠĠĠ Ġ. map Concurrent ly ( this , Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġevents . add ( it ) ĊĠĠĠĠĠĠĠĠĠ Ġ- it ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ. toList () ĊĠĠĠ ĠassertThat ( events ). contains Exactly ( 1 , Ġ 2 ). in Order () ĊĠĠĠ ĠassertThat ( result ). contains Exactly (- 1 , Ġ- 2 ). in Order () ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Ġreturns Ġnull Ġfor Ġempty Ġflow `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ ĠassertThat ( flow Of < String >(). consume First ()). isNull () ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Ġhas Remaining Ġis Ġfalse Ġfor Ġsingle - item Ġflow `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ ĠassertNotNull ( flow Of (" foo "). consume First ()). use Ġ{ Ġconsumed Ġ-> ĊĠĠĠĠĠ ĠassertFalse ( consumed . has Remaining ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Ġremaining Ġis Ġempty Ġfor Ġsingle - item Ġflow `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ ĠassertNotNull ( flow Of (" foo "). consume First ()). use Ġ{ Ġconsumed Ġ-> ĊĠĠĠĠĠ ĠassertThat ( consumed . remaining . toList ()). isEmpty () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Ġitem Ġis Ġfirst Ġitem Ġfor Ġsingle - item Ġflow `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġitem Ġ= Ġ" foo " ĊĠĠĠ ĠassertNotNull ( flow Of ( item ). consume First ()). use Ġ{ Ġconsumed Ġ-> ĊĠĠĠĠĠ ĠassertThat ( consumed . item ). isEqualTo ( item ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Ġhas Remaining Ġis Ġtrue Ġfor Ġmulti - item Ġflow `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġitems Ġ= ĠlistOf (" foo ", Ġ" bar ", Ġ" baz ") ĊĠĠĠ ĠassertNotNull ( items . as Flow (). consume First ()). use Ġ{ Ġconsumed Ġ-> ĊĠĠĠĠĠ ĠassertTrue ( consumed . has Remaining ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Ġitem Ġis Ġfirst Ġitem Ġfor Ġmulti - item Ġflow `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġitems Ġ= ĠlistOf (" foo ", Ġ" bar ", Ġ" baz ") ĊĠĠĠ ĠassertNotNull ( items . as Flow (). consume First ()). use Ġ{ Ġconsumed Ġ-> ĊĠĠĠĠĠ ĠassertThat ( consumed . item ). isEqualTo ( items . first ()) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Ġremaining Ġcontains Ġremaining Ġitems `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġitems Ġ= ĠlistOf (" foo ", Ġ" bar ", Ġ" baz ") ĊĠĠĠ ĠassertNotNull ( items . as Flow (). consume First ()). use Ġ{ Ġconsumed Ġ-> ĊĠĠĠĠĠ ĠassertThat ( consumed . remaining . toList ()). contains Exactly Elements In ( items . drop ( 1 )). in Order () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Or Ġreturns Ġalternate Ġfor Ġempty Ġflow `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġalternate Ġ= Ġ" foo " ĊĠĠĠ Ġflow Of < String >(). consume First Or Ġ{ Ġalternate Ġ}. use Ġ{ Ġconsumed Ġ-> ĊĠĠĠĠĠ ĠassertThat ( consumed . item ). isEqualTo ( alternate ) ĊĠĠĠĠĠ ĠassertFalse ( consumed . has Remaining ) ĊĠĠĠĠĠ ĠassertThat ( consumed . remaining . toList ()). isEmpty () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Test ĊĠ Ġfun Ġ` consume First Or Ġreturns Ġsame Ġas Ġconsume First Ġfor Ġnon - empty Ġflow `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġitems Ġ= ĠlistOf (" foo ", Ġ" bar ", Ġ" baz ") ĊĠĠĠ Ġitems . as Flow (). consume First Or Ġ{ Ġ" alternate " Ġ}. use Ġ{ Ġconsumed 1 Ġ-> ĊĠĠĠĠĠ ĠassertNotNull ( items . as Flow (). consume First ()). use Ġ{ Ġconsumed 2 Ġ-> ĊĠĠĠĠĠĠĠ ĠassertThat ( consumed 1 . item ). isEqualTo ( consumed 2 . item ) ĊĠĠĠĠĠĠĠ ĠassertThat ( consumed 1 . has Remaining ). isEqualTo ( consumed 2 . has Remaining ) ĊĠĠĠĠĠĠĠ ĠassertThat ( consumed 1 . remaining . toList ()) ĊĠĠĠĠĠĠĠĠĠ Ġ. contains Exactly Elements In ( consumed 2 . remaining . toList ()) ĊĠĠĠĠĠĠĠĠĠ Ġ. in Order () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ