/* Ċ Ġ* ĠCopyright Ġ 2 0 1 6 - 2 0 1 9 ĠJetBrains Ġs . r . o . ĠUse Ġof Ġthis Ġsource Ġcode Ġis Ġgoverned Ġby Ġthe ĠApache Ġ 2 . 0 Ġlicense . Ċ Ġ*/ Ċ Ċ @ file : Suppress (" UN CHECK ED _ CAST ") Ċ Ċ package Ġkotlinx . coroutines . reactive . flow Ċ Ċ import Ġkotlinx . coroutines . flow .* Ċ import Ġorg . junit .* Ċ import Ġorg . reactiv est ream s .* Ċ import Ġorg . reactiv est ream s . t ck .* Ċ Ċ import Ġorg . junit . Assert .* Ċ import Ġorg . reactiv est ream s . Subscription Ċ import Ġorg . reactiv est ream s . Subscriber Ċ import Ġjava . util . ArrayList Ċ import Ġjava . util . concurrent .* Ċ import Ġjava . util . concurrent . Count DownLatch Ċ import Ġjava . util . concurrent . Fork Join Pool . common Pool Ċ Ċ class ĠIterable Flow T ck Test Ġ: ĠPublisher Verification < Long >( Test Environment ()) Ġ{ ĊĊĠĠĠ Ġprivate Ġfun Ġgenerate ( num : ĠLong ): ĠArray < Long > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠArray ( if Ġ( num Ġ>= ĠInteger . MAX _ VALUE ) Ġ 1 0 0 0 0 0 0 Ġelse Ġnum . toInt ()) Ġ{ Ġit . to Long () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġcreate Publisher ( elements : ĠLong ): ĠPublisher < Long > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġgenerate ( elements ). as Iterable (). as Flow (). as Publisher () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress (" Subscriber Implementation ") ĊĠĠĠ Ġoverride Ġfun Ġcreate Failed Publisher (): ĠPublisher < Long >? Ġ{ ĊĠĠĠĠĠĠĠ Ġ/* ĊĠĠĠĠĠĠĠĠ Ġ* ĠThis Ġis Ġa Ġhack Ġfor Ġour Ġadapter Ġstructure : ĊĠĠĠĠĠĠĠĠ Ġ* ĠTests Ġassume Ġthat Ġcalling Ġ" collect " Ġis Ġenough Ġfor Ġpublisher Ġto Ġfail Ġand Ġit Ġis Ġnot ĊĠĠĠĠĠĠĠĠ Ġ* Ġtrue Ġfor Ġour Ġimplementation ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġval Ġpub Ġ= Ġ{ Ġerror ( 4 2 ) Ġ}. as Flow (). as Publisher () ĊĠĠĠĠĠĠĠ Ġreturn ĠPublisher Ġ{ Ġsubscriber Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpub . subscribe ( object Ġ: ĠSubscriber < Long > Ġby Ġsubscriber Ġas ĠSubscriber < Long > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Subscribe ( s : ĠSubscription ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . on Subscribe ( s ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . request ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Stack Overflow Tr amp oline () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġl atch Ġ= ĠCount DownLatch ( 1 ) ĊĠĠĠĠĠĠĠ Ġval Ġcollected Ġ= ĠArrayList < Long >() ĊĠĠĠĠĠĠĠ Ġval Ġto Request Ġ= Ġ 1 0 0 0 L ĊĠĠĠĠĠĠĠ Ġval Ġarray Ġ= Ġgenerate ( to Request ) ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= Ġarray . as Iterable (). as Flow (). as Publisher () ĊĊĠĠĠĠĠĠĠ Ġpublisher . subscribe ( object Ġ: ĠSubscriber < Long > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġlateinit Ġvar Ġs : ĠSubscription ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Subscribe ( s : ĠSubscription ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . s Ġ= Ġs ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . request ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( a Long : ĠLong ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcollected . add ( a Long ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs . request ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonError ( t : ĠThrowable ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġl atch . count Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}) ĊĊĠĠĠĠĠĠĠ Ġl atch . await ( 5 , ĠTimeUnit . SECONDS ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( collected , Ġarray . toList ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Concurrent Request () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġl atch Ġ= ĠCount DownLatch ( 1 ) ĊĠĠĠĠĠĠĠ Ġval Ġcollected Ġ= ĠArrayList < Long >() ĊĠĠĠĠĠĠĠ Ġval Ġn Ġ= Ġ 5 0 0 0 0 L ĊĠĠĠĠĠĠĠ Ġval Ġarray Ġ= Ġgenerate ( n ) ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= Ġarray . as Iterable (). as Flow (). as Publisher () ĊĊĠĠĠĠĠĠĠ Ġpublisher . subscribe ( object Ġ: ĠSubscriber < Long > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġvar Ġs : ĠSubscription ? Ġ= Ġnull ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Subscribe ( s : ĠSubscription ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . s Ġ= Ġs ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ( i Ġin Ġ 0 Ġuntil Ġn ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommon Pool (). execute Ġ{ Ġs . request ( 1 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( a Long : ĠLong ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcollected . add ( a Long ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonError ( t : ĠThrowable ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġl atch . count Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}) ĊĊĠĠĠĠĠĠĠ Ġl atch . await ( 5 0 , ĠTimeUnit . SECONDS ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( array . toList (), Ġcollected ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Ignore ĊĠĠĠ Ġoverride Ġfun Ġrequired _ spec 3 0 9 _ request Zero Must Signal IllegalArgumentException () Ġ{ ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Ignore ĊĠĠĠ Ġoverride Ġfun Ġrequired _ spec 3 0 9 _ request Negative Number Must Signal IllegalArgumentException () Ġ{ ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Ignore ĊĠĠĠ Ġoverride Ġfun Ġrequired _ spec 3 1 2 _ cancel Must Make The Publisher To Event ually Stop Signal ing () Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠThis Ġtest Ġhas Ġa Ġbug Ġin Ġit ĊĠĠĠ Ġ} Ċ } Ċ