/* Ċ Ġ* ĠCopyright Ġ 2 0 2 0 ĠThe ĠAndroid ĠOpen Ġstream ĠProject Ċ Ġ* Ċ Ġ* Ġ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 Ġandroidx . compose . runtime . rx java 2 Ċ Ċ import Ġandroidx . compose . runtime . Composable Ċ import Ġandroidx . compose . runtime . State Ċ import Ġandroidx . compose . runtime . getValue Ċ import Ġandroidx . compose . runtime . mutable State Of Ċ import Ġandroidx . compose . runtime . setValue Ċ import Ġandroidx . test . filters . Medium Test Ċ import Ġandroidx . ui . test . create Compose Rule Ċ import Ġcom . google . common . truth . Truth . assertThat Ċ import Ġio . reactivex . Back pressure Strategy Ċ import Ġio . reactivex . Maybe Ċ import Ġio . reactivex . Maybe Emitter Ċ import Ġio . reactivex . Maybe On Subscribe Ċ import Ġio . reactivex . Single Ċ import Ġio . reactivex . Single Emitter Ċ import Ġio . reactivex . Single On Subscribe Ċ import Ġio . reactivex . subjects . Behavior Subject Ċ import Ġorg . junit . Rule Ċ import Ġorg . junit . Test Ċ import Ġorg . junit . runner . RunWith Ċ import Ġorg . junit . runners . Parameterized Ċ import Ġjava . util . concurrent . Count DownLatch Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ @ Medium Test Ċ @ RunWith ( Parameterized :: class ) Ċ class ĠRx Java 2 Adapter Test ( private Ġval Ġfactory : Ġ() Ġ-> ĠStream ) Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġ@ Parameterized . Parameters ( name Ġ= Ġ"{ 0 }") ĊĠĠĠĠĠĠĠ Ġfun Ġinit Parameters (): ĠArray < () Ġ-> ĠStream > Ġ= ĠarrayOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠObservable Stream (), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠFlow able Stream (), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSingle Stream (), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMaybe Stream () ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ get : Rule ĊĠĠĠ Ġval Ġrule Ġ= Ġcreate Compose Rule () ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġwhen Value IsNot Set We Got Null () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= Ġfactory () ĊĠĠĠĠĠĠĠ Ġvar Ġreal Value : ĠString ? Ġ= Ġ" to - be - updated " ĊĠĠĠĠĠĠĠ Ġrule . setContent Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreal Value Ġ= Ġstream . subscribe As State ( null ). value ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ ĠassertThat ( real Value ). isNull () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġwe Got Initial Value () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= Ġfactory () ĊĠĠĠĠĠĠĠ Ġstream . onNext (" value ") ĊĠĠĠĠĠĠĠ Ġvar Ġreal Value : ĠString ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġrule . setContent Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreal Value Ġ= Ġstream . subscribe As State ( null ). value ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ ĠassertThat ( real Value ). isEqualTo (" value ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġwe Receive Value Submitted Later () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= Ġfactory () ĊĊĠĠĠĠĠĠĠ Ġvar Ġreal Value : ĠString ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġrule . setContent Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreal Value Ġ= Ġstream . subscribe As State ( null ). value ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġrule . run On Idle Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . onNext (" value ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġrule . run On Idle Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( real Value ). isEqualTo (" value ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġwe Receive Second Value () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= Ġfactory () ĊĠĠĠĠĠĠĠ Ġif Ġ(! stream . support Multiple Values ()) Ġreturn ĊĊĠĠĠĠĠĠĠ Ġstream . onNext (" value ") ĊĠĠĠĠĠĠĠ Ġvar Ġreal Value : ĠString ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġrule . setContent Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreal Value Ġ= Ġstream . subscribe As State ( null ). value ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġrule . run On Idle Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . onNext (" value 2 ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġrule . run On Idle Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( real Value ). isEqualTo (" value 2 ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġno Updates After Dispose () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= Ġfactory () ĊĠĠĠĠĠĠĠ Ġvar Ġemit Ġby Ġmutable State Of ( true ) ĊĠĠĠĠĠĠĠ Ġvar Ġreal Value : ĠString ? Ġ= Ġ" to - be - updated " ĊĠĠĠĠĠĠĠ Ġrule . setContent Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( emit ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreal Value Ġ= Ġstream . subscribe As State ( null ). value ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġrule . run On Idle Ġ{ Ġemit Ġ= Ġfalse Ġ} ĊĊĠĠĠĠĠĠĠ Ġrule . run On Idle Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . onNext (" value ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġrule . run On Idle Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( real Value ). isNull () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Observable With Initial Value () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= Ġfactory () ĊĠĠĠĠĠĠĠ Ġvar Ġreal Value : ĠString ? Ġ= Ġ" to - be - updated " ĊĠĠĠĠĠĠĠ Ġrule . setContent Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreal Value Ġ= Ġstream . subscribe As State (" value "). value ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ ĠassertThat ( real Value ). isEqualTo (" value ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġwe Receive Value Submitted On Background Thread () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= Ġfactory () ĊĠĠĠĠĠĠĠ Ġval Ġl atch Ġ= ĠCount DownLatch ( 1 ) ĊĊĠĠĠĠĠĠĠ Ġvar Ġreal Value : ĠString ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġrule . setContent Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreal Value Ġ= Ġstream . subscribe As State ( null ). value ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( real Value Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġl atch . count Down () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ ĠThread ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRunnable Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream . onNext (" value ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ). start () ĊĊĠĠĠĠĠĠĠ ĠassertThat ( latch . await ( 5 , ĠTimeUnit . SECONDS )). isTrue () ĊĊĠĠĠĠĠĠĠ Ġrule . run On Ui Thread Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( real Value ). isEqualTo (" value ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ interface ĠStream Ġ{ ĊĠĠĠ Ġfun Ġon Next ( value : ĠString ) ĊĊĠĠĠ Ġ@ Composable ĊĠĠĠ Ġfun Ġsubscribe As State ( initial : ĠString ?): ĠState < String ?> ĊĊĠĠĠ Ġfun Ġsupport Multiple Values (): ĠBoolean Ċ } Ċ Ċ private Ġclass ĠObservable Stream Ġ: Ġ() Ġ-> ĠStream Ġ{ ĊĊĠĠĠ Ġoverride Ġfun Ġinvoke (): ĠStream Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġobject Ġ: ĠStream Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsubject Ġ= ĠBehavior Subject . create < String >() ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( value : ĠString ) Ġ= Ġsubject . onNext ( value ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Composable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsubscribe As State ( initial : ĠString ?) Ġ= Ġsubject . subscribe As State ( initial ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsupport Multiple Values (): ĠBoolean Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠtoString () Ġ= Ġ" Observable " Ċ } Ċ Ċ private Ġclass ĠFlow able Stream Ġ: Ġ() Ġ-> ĠStream Ġ{ ĊĠĠĠ Ġoverride Ġfun Ġinvoke (): ĠStream Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġobject Ġ: ĠStream Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsubject Ġ= ĠBehavior Subject . create < String >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġflow able Ġ= Ġsubject . to Flow able ( Back pressure Strategy . LATE ST ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( value : ĠString ) Ġ= Ġsubject . onNext ( value ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Composable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsubscribe As State ( initial : ĠString ?) Ġ= Ġflow able . subscribe As State ( initial ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsupport Multiple Values (): ĠBoolean Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠtoString () Ġ= Ġ" Flow able " Ċ } Ċ Ċ private Ġclass ĠSingle Stream Ġ: Ġ() Ġ-> ĠStream Ġ{ ĊĠĠĠ Ġoverride Ġfun Ġinvoke (): ĠStream Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġobject Ġ: ĠStream , ĠSingle On Subscribe < String > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġvalue : ĠString ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġem mit ers Ġ= Ġmutable ListOf < Single Emitter < String >>() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsingle Ġ= ĠSingle . create ( this ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( value : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( this . value Ġ== Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . value Ġ= Ġvalue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit All () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġfun Ġemit All () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( value Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem mit ers . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! it . is Disposed ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . on Success ( value !! ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsubscribe ( p 0 : ĠSingle Emitter < String >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem mit ers . add ( p 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit All () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Composable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsubscribe As State ( initial : ĠString ?) Ġ= Ġsingle . subscribe As State ( initial ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsupport Multiple Values (): ĠBoolean Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠtoString () Ġ= Ġ" Single " Ċ } Ċ Ċ private Ġclass ĠMaybe Stream Ġ: Ġ() Ġ-> ĠStream Ġ{ ĊĠĠĠ Ġoverride Ġfun Ġinvoke (): ĠStream Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġobject Ġ: ĠStream , ĠMaybe On Subscribe < String > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġem mit ers Ġ= Ġmutable ListOf < Maybe Emitter < String >>() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġvalue : ĠString ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmaybe Ġ= ĠMaybe . create < String >( this ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( value : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( this . value Ġ== Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . value Ġ= Ġvalue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit All () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġfun Ġemit All () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( value Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem mit ers . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! it . is Disposed ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . on Success ( value !! ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsubscribe ( p 0 : ĠMaybe Emitter < String >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem mit ers . add ( p 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit All () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Composable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsubscribe As State ( initial : ĠString ?) Ġ= Ġmaybe . subscribe As State ( initial ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsupport Multiple Values (): ĠBoolean Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠtoString () Ġ= Ġ" Maybe " Ċ }