package Ġcom . h hd . my and vide ot est . my test Ċ Ċ import Ġandroid . annotation . Suppress Lint Ċ import Ġandroid . content . Context Ċ import Ġandroid . os . Bundle Ċ import Ġandroid . widget . Button Ċ import Ġandroidx . appcompat . app . AppCompatActivity Ċ import Ġcom . h hd . my and vide ot est . util . Log Ex Ċ import Ġcom . h hd . my and vide ot est . util . My Activity Util Ċ import Ġcom . h hd . my and vide ot est . util . My Util Ċ import Ġio . reactivex .* Ċ import Ġio . reactivex . Observable Ċ import Ġio . reactivex . Observer Ċ import Ġio . reactivex . android . schedul ers . Android Schedulers Ċ import Ġio . reactivex . dis pos ables . Disposable Ċ import Ġio . reactivex . functions . Bi Function Ċ import Ġio . reactivex . processors . Publish Processor Ċ import Ġio . reactivex . rx kotlin . subscribe By Ċ import Ġio . reactivex . rx kotlin . to Flow able Ċ import Ġio . reactivex . rx kotlin . to Observable Ċ import Ġio . reactivex . schedul ers . Schedulers Ċ import Ġio . reactivex . subjects . Async Subject Ċ import Ġio . reactivex . subjects . Behavior Subject Ċ import Ġio . reactivex . subjects . Publish Subject Ċ import Ġio . reactivex . subjects . Replay Subject Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġorg . reactiv est ream s . Subscriber Ċ import Ġorg . reactiv est ream s . Subscription Ċ import Ġjava . nio . ByteBuffer Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . Executors Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġjava . util . concurrent . locks . Re entrant Lock Ċ import Ġkotlin . concurrent . with Lock Ċ Ċ class ĠMy Test Activity Ġ: ĠAppCompatActivity () Ġ{ ĊĠĠĠ Ġoverride Ġfun ĠonCreate ( savedInstanceState : ĠBundle ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġsuper . onCreate ( savedInstanceState ) ĊĊĠĠĠĠĠĠĠ ĠMy Activity Util . set ActionBar Hide ( this ) ĊĠĠĠĠĠĠĠ ĠMy Activity Util . set Keep Screen On ( this ) ĊĠĠĠĠĠĠĠ Ġval Ġcontext Ġ= Ġthis Ġas ĠContext ĊĠĠĠĠĠĠĠ Ġval Ġthis Obj Ġ= Ġthis ĊĠĠĠĠĠĠĠ Ġval Ġsv Ġ= ĠMy Util . create ScrollView Mp Wc ( context ) ĊĠĠĠĠĠĠĠ Ġval Ġfl Ġ= ĠMy Util . create Flex box Layout Mp Wc ( context ) ĊĠĠĠĠĠĠĠ Ġthis . set ContentView ( sv ) ĊĠĠĠĠĠĠĠ Ġsv . addView ( fl ) ĊĊĠĠĠĠĠĠĠ Ġthis . java Class . methods ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . name . contains ("_ ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Not Ġ{ Ġit . name . contains ("$ ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmethod Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbtn Ġ= ĠButton ( context ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfl . addView ( btn ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbtn . is All Caps Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbtn . text Ġ= Ġmethod . name ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbtn . setOnClickListener Ġ{ Ġmethod . invoke ( this Obj ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġ_ 0 1 _ Observable _ subscribe By () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob Ġ= ĠlistOf (" one ", Ġ" two ", Ġ" three "). to Observable () ĊĠĠĠĠĠĠĠ Ġob . subscribe By ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĠLog Ex . d (" onNext Ġit [$ it ]") Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĠLog Ex . d (" onError Ġit [$ it ]") Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ ĠonComplete Ġ= Ġ{ ĠLog Ex . d (" onComplete ") Ġ} ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 0 2 _ Publish Subject _ map _ subscribe () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsb Ġ= ĠPublish Subject . create < Int >() ĊĊĠĠĠĠĠĠĠ Ġsb . map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" map Ġit [$ it ]") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit Ġ% Ġ 2 Ġ== Ġ 0 ĊĠĠĠĠĠĠĠ Ġ}. subscribe ({ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġit [$ it ]") ĊĠĠĠĠĠĠĠ Ġ}) ĊĊĠĠĠĠĠĠĠ Ġsb . onNext ( 4 ) ĊĠĠĠĠĠĠĠ Ġsb . onNext ( 9 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 0 3 _ Maybe _ just () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmb Ġ= ĠMaybe . just ( 1 4 ) ĊĠĠĠĠĠĠĠ Ġmb . subscribe By ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠonSuccess Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" on Success Ġit [$ it ]") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" onError Ġit [$ it ]") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ ĠonComplete Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" onComplete ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass Ġ_ My Basic Observer Ġ: ĠObserver < Any > Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ this . hashCode ()} ĠonComplete ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Subscribe ( d : ĠDis posable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ this . hashCode ()} Ġon Subscribe Ġd [$ d ]") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( t : ĠAny ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ this . hashCode ()} Ġon Next Ġt [$ t ]") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonError ( e : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ this . hashCode ()} ĠonError Ġe [$ e ]") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 0 4 _ Observable _ subscribe () Ġ{ ĊĠĠĠĠĠĠĠ ĠlistOf ( 1 , Ġ 2 , Ġ 3 , Ġ 4 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Observable () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe (_ My Basic Observer ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 0 5 _ Observable _ just () Ġ{ ĊĠĠĠĠĠĠĠ ĠObservable . just ( 1 ). subscribe (_ My Basic Observer ()) ĊĠĠĠĠĠĠĠ ĠObservable . just ( 1 0 , Ġ 2 0 , Ġ 3 0 ). subscribe (_ My Basic Observer ()) ĊĠĠĠĠĠĠĠ ĠObservable . just ( listOf ( 1 0 0 , Ġ 2 0 0 , Ġ 3 0 0 )). subscribe (_ My Basic Observer ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 0 6 _ Observable _ interval () Ġ{ ĊĠĠĠĠĠĠĠ ĠObservable . interval ( 1 0 0 , ĠTimeUnit . MILLISECONDS ). take While Ġ{ Ġit Ġ< Ġ 1 0 0 Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe (_ My Basic Observer ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 0 7 _ hot _ Observable _ Connect able Observable _ publish _ connect () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġc ob Ġ= ĠlistOf ( 1 , Ġ 2 , Ġ 3 ). to Observable (). publish () ĊĠĠĠĠĠĠĠ Ġc ob . subscribe (_ My Basic Observer ()) ĊĠĠĠĠĠĠĠ Ġc ob . connect () ĊĠĠĠĠĠĠĠ Ġc ob . subscribe (_ My Basic Observer ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 0 8 _ Publish Subject _ mid _ subscribe () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob Ġ= ĠObservable . interval ( 1 0 0 , ĠTimeUnit . MILLISECONDS ). take While Ġ{ Ġit Ġ< Ġ 5 0 Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġps Ġ= ĠPublish Subject . create < Long >() ĊĠĠĠĠĠĠĠ Ġob . subscribe ( ps ) ĊĠĠĠĠĠĠĠ Ġps . subscribe (_ My Basic Observer ()) ĊĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 2 _ 0 0 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġps . subscribe (_ My Basic Observer ()) ĊĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 2 _ 0 0 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 0 9 _ Async Subject _ basic () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob Ġ= ĠObservable . just ( 1 , Ġ 2 , Ġ 3 ) ĊĠĠĠĠĠĠĠ Ġval Ġa Sub Ġ= ĠAsync Subject . create < Int >() ĊĠĠĠĠĠĠĠ Ġob . subscribe ( a Sub ) ĊĠĠĠĠĠĠĠ Ġa Sub . subscribe (_ My Basic Observer ()) ĊĠĠĠĠĠĠĠ Ġa Sub . onComplete () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 0 _ Async Subject _ 2 _ Observer () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġa Sub Ġ= ĠAsync Subject . create < Int >() ĊĠĠĠĠĠĠĠ Ġa Sub . onNext ( 1 ) ĊĠĠĠĠĠĠĠ Ġa Sub . onNext ( 2 ) ĊĠĠĠĠĠĠĠ Ġa Sub . onNext ( 3 ) ĊĠĠĠĠĠĠĠ Ġa Sub . subscribe (_ My Basic Observer ()) ĊĊĠĠĠĠĠĠĠ Ġa Sub . onNext ( 4 ) ĊĠĠĠĠĠĠĠ Ġa Sub . onNext ( 5 ) ĊĠĠĠĠĠĠĠ Ġa Sub . onNext ( 6 ) ĊĠĠĠĠĠĠĠ Ġa Sub . subscribe (_ My Basic Observer ()) ĊĊĠĠĠĠĠĠĠ Ġa Sub . onComplete () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 1 _ Publish Subject _ basic () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġp Sub Ġ= ĠPublish Subject . create < Int >() ĊĠĠĠĠĠĠĠ Ġp Sub . subscribe (_ My Basic Observer ()) ĊĠĠĠĠĠĠĠ Ġp Sub . onNext ( 1 ) ĊĠĠĠĠĠĠĠ Ġp Sub . onNext ( 2 ) ĊĠĠĠĠĠĠĠ Ġp Sub . onNext ( 3 ) ĊĊĠĠĠĠĠĠĠ Ġp Sub . subscribe (_ My Basic Observer ()) ĊĠĠĠĠĠĠĠ Ġp Sub . onNext ( 4 ) ĊĠĠĠĠĠĠĠ Ġp Sub . onNext ( 5 ) ĊĠĠĠĠĠĠĠ Ġp Sub . onNext ( 6 ) ĊĊĠĠĠĠĠĠĠ Ġp Sub . onComplete () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 2 _ Behavior Subject _ very _ last () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġb Sub Ġ= ĠBehavior Subject . create < Int >() ĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 1 ) ĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 2 ) ĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 3 ) ĊĠĠĠĠĠĠĠ Ġb Sub . subscribe (_ My Basic Observer ()) ĊĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 4 ) ĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 5 ) ĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 6 ) ĊĠĠĠĠĠĠĠ Ġb Sub . subscribe (_ My Basic Observer ()) ĊĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 7 ) ĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 8 ) ĊĠĠĠĠĠĠĠ Ġb Sub . onNext ( 9 ) ĊĠĠĠĠĠĠĠ Ġb Sub . onComplete () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 3 _ Replay Subject _ all () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġr Sub Ġ= ĠRe play Subject . create < Int >() ĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 1 ) ĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 2 ) ĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 3 ) ĊĠĠĠĠĠĠĠ Ġr Sub . subscribe (_ My Basic Observer ()) ĊĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 4 ) ĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 5 ) ĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 6 ) ĊĠĠĠĠĠĠĠ Ġr Sub . subscribe (_ My Basic Observer ()) ĊĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 7 ) ĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 8 ) ĊĠĠĠĠĠĠĠ Ġr Sub . onNext ( 9 ) ĊĠĠĠĠĠĠĠ Ġr Sub . onComplete () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 4 _ Behavior Subject _ observe On _ computation _ bug () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob Ġ= ĠObservable . just ( 1 , Ġ 2 , Ġ 3 ) ĊĠĠĠĠĠĠĠ Ġval Ġb Sub Ġ= ĠBehavior Subject . create < Int >() ĊĊĠĠĠĠĠĠĠ Ġb Sub . observe On ( Schedulers . computation ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ hashCode ()} Ġsubscribe Ġit [$ it ] Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ hashCode ()} Ġsubscribe Ġit [$ it ] Ġend ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġb Sub . observe On ( Schedulers . computation ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ hashCode ()} Ġsubscribe Ġit [$ it ] Ġreceived !! !") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġob . subscribe ( b Sub ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 5 _ Behavior Subject _ observe On _ computation _ bug _ 2 () Ġ{ ĊĠĠĠĠĠĠĠ ĠObservable . range ( 0 , Ġ 1 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" map Ġ${ it }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . computation ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 6 _ Flow able _ basic () Ġ{ ĊĠĠĠĠĠĠĠ ĠFlow able . range ( 0 , Ġ 1 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" map Ġ${ it }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . toString () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 7 _ Flow able _ Subscription _ request () Ġ{ ĊĠĠĠĠĠĠĠ ĠFlow able . range ( 1 , Ġ 1 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( object Ġ: ĠSubscriber < Int > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" onComplete ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Subscribe ( s : ĠSubscription ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs !!. request ( 5 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( t : ĠInt ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" onNext Ġ[$ t ]") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonError ( t : ĠThrowable ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġ_ run _ Flow able ( st : ĠBack pressure Strategy ) Ġ{ ĊĠĠĠĠĠĠĠ ĠObservable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. range ( 1 , Ġ 1 _ 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Flow able ( st ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" map Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . toString () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe By ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" onNext Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" onNext Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . e (" onError Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 8 _ Back pressure Strategy _ BUFFER () Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ run _ Flow able ( Back pressure Strategy . BUFFER ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 1 9 _ Back pressure Strategy _ ERROR () Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ run _ Flow able ( Back pressure Strategy . ERROR ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 0 _ Back pressure Strategy _ LATE ST () Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ run _ Flow able ( Back pressure Strategy . LATE ST ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 1 _ Back pressure Strategy _ DROP () Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ run _ Flow able ( Back pressure Strategy . DROP ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Suppress Lint (" Check Result ") ĊĠĠĠ Ġfun Ġ_ 2 2 _ Back pressure Strategy _ MISSING _ buf _ 1 0 () Ġ{ ĊĠĠĠĠĠĠĠ Ġ( 1 .. 1 _ 0 0 0 ). toList (). to Flow able () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Back pressure Buffer ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 1 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ ĠLog Ex . d (" overflow ") Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠBack pressure Overflow Strategy . DROP _ OL DEST ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Back pressure Drop () Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Back pressure Latest () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . computation (), Ġfalse , Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" map Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . toString () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . computation ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe By ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" onNext Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" onNext Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . e (" onError Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 3 _ Connect able Flow able _ publish _ connect () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcon Fw Ġ= ĠlistOf ( 1 , Ġ 2 , Ġ 3 ). to Flow able (). publish () ĊĊĠĠĠĠĠĠĠ Ġcon Fw . subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 1 st Ġsubscribe Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 1 st Ġsubscribe Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcon Fw . subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 2 nd Ġsubscribe Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 2 nd Ġsubscribe Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcon Fw . subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 3 rd Ġsubscribe Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 3 rd Ġsubscribe Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcon Fw . connect () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 4 _ Flow able _ Publish Processor _ subscribe () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfw Ġ= ĠlistOf ( 1 , Ġ 2 , Ġ 3 ). to Flow able () ĊĠĠĠĠĠĠĠ Ġval Ġpp Ġ= ĠPublish Processor . create < Int >() ĊĊĠĠĠĠĠĠĠ Ġpp . subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 1 st Ġsubscribe Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 1 st Ġsubscribe Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġpp . subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 2 nd Ġsubscribe Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 2 nd Ġsubscribe Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġpp . subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 3 rd Ġsubscribe Ġ$ it Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" 3 rd Ġsubscribe Ġ$ it Ġend ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfw . subscribe ( pp ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 5 _ Flow able _ buffer _ bound () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbound Fw Ġ= ĠFlow able . interval ( 3 5 0 , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠ Ġval Ġsrc Fw Ġ= ĠFlow able . interval ( 1 0 0 , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠ Ġsrc Fw . buffer ( bound Fw ). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġit [$ it ]") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 6 _ Flow able _ throttle First () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfw Ġ= ĠFlow able . interval ( 1 0 0 , ĠTimeUnit . MILLISECONDS ) ĊĊĠĠĠĠĠĠĠ Ġfw . throttle First ( 1 _ 0 0 0 , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġit [$ it ]") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 7 _ de bounce () Ġ{ ĊĠĠĠĠĠĠĠ ĠObservable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create < Int > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . onNext ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . onNext ( 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . onNext ( 3 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . onComplete () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. de bounce ( 5 _ 0 0 0 , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 8 _ Flow able _ filter () Ġ{ ĊĠĠĠĠĠĠĠ Ġ( 1 .. 1 0 0 ). toList (). to Flow able (). filter Ġ{ Ġit Ġ% Ġ 1 0 Ġ== Ġ 0 Ġ}. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 2 9 _ Observable _ flatMap () Ġ{ ĊĠĠĠĠĠĠĠ Ġ( 0 .. 1 0 ). toList (). to Observable (). flatMap Ġ{ Ġnum Ġ-> ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġob Ġ= ĠarrayOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnum Ġ* Ġ 1 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnum Ġ* Ġ 1 0 Ġ+ Ġ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnum Ġ* Ġ 1 0 Ġ+ Ġ 2 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ). to Observable () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ flatMap Ġob ĊĠĠĠĠĠĠĠ Ġ}. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 3 0 _ Observable _ reduce () Ġ{ ĊĠĠĠĠĠĠĠ Ġ( 0 .. 1 0 ). toList (). to Observable (). reduce Ġ{ Ġaccum , ĠnewValue Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġres Ġ= Ġaccum Ġ+ ĠnewValue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġres ĊĠĠĠĠĠĠĠ Ġ}. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 3 1 _ Observable _ zip () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob 1 Ġ= ĠObservable . interval ( 1 0 0 , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠ Ġval Ġob 2 Ġ= ĠObservable . interval ( 2 5 0 , ĠTimeUnit . MILLISECONDS ) ĊĊĠĠĠĠĠĠĠ ĠObservable . zip ( ob 1 , Ġob 2 , ĠBi Function Ġ{ Ġt 1 : ĠLong , Ġt 2 : ĠLong Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"$ t 1 Ġ$ t 2 " ĊĠĠĠĠĠĠĠ Ġ}). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 3 2 _ Observable _ combine Latest () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob 1 Ġ= ĠObservable . interval ( 1 0 0 , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠ Ġval Ġob 2 Ġ= ĠObservable . interval ( 2 5 0 , ĠTimeUnit . MILLISECONDS ) ĊĊĠĠĠĠĠĠĠ ĠObservable . combine Latest ( ob 1 , Ġob 2 , ĠBi Function Ġ{ Ġt 1 : ĠLong , Ġt 2 : ĠLong Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"$ t 1 Ġ$ t 2 " ĊĠĠĠĠĠĠĠ Ġ}). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 3 3 _ Observable _ merge () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob 1 Ġ= Ġ( 0 .. 1 0 ). toList (). to Observable () ĊĠĠĠĠĠĠĠ Ġval Ġob 2 Ġ= Ġ( 1 0 0 .. 1 1 0 ). toList (). to Observable () ĊĊĠĠĠĠĠĠĠ ĠObservable . merge ( ob 1 , Ġob 2 ). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 3 4 _ Observable _ amb () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob 1 Ġ= ĠObservable . interval ( 1 0 0 , ĠTimeUnit . MILLISECONDS ). map Ġ{ Ġ" ob 1 Ġ$ it " Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġob 2 Ġ= ĠObservable . interval ( 5 0 0 , ĠTimeUnit . MILLISECONDS ). map Ġ{ Ġ" ob 2 Ġ$ it " Ġ} ĊĊĠĠĠĠĠĠĠ ĠObservable . amb ( listOf ( ob 1 , Ġob 2 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 3 5 _ video _ frame _ em ul () Ġ{ ĊĊĠĠĠĠĠĠĠ ĠFlow able ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create < Int >({ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġem Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ( 0 .. 1 0 ). forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ Thread . currentThread (). name } Ġsrc Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . onNext ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . onComplete () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĠBack pressure Strategy . BUFFER ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe On ( Schedulers . new Thread ()) Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . new Thread (), Ġfalse , Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 5 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ Thread . currentThread (). name } Ġmap Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnew Str Ġ= Ġ"$ it Ġ$ it " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnew Str ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . new Thread (), Ġfalse , Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 7 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ Thread . currentThread (). name } Ġmap 2 Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnew Str Ġ= Ġ"$ it Ġ$ it " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnew Str ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . new Thread (), Ġfalse , Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("${ Thread . currentThread (). name } Ġsubscribe 1 Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 3 6 _ Observable _ flatMap _ new Thread () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġ( 1 .. 1 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. toList () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Observable () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . new Thread ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( Random (). nextInt ( 3 ). to Long () Ġ* Ġ 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" flatMap Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnum Ġ= Ġit ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠObservable . create < Int > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġem Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ( 0 .. 1 0 ). forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnew Num Ġ= Ġnum Ġ* Ġ 1 0 Ġ+ Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" flatMap Ġon Next Ġ$ new Num ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . onNext ( new Num ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġem . onComplete () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . new Thread ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" subscribe Ġ$ it ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġ_ 3 7 _ sequence Equal () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġob 1 Ġ= ĠarrayOf ( 1 , Ġ 2 , Ġ 3 ). to Observable () ĊĠĠĠĠĠĠĠ Ġval Ġob 2 Ġ= ĠarrayOf ( 1 , Ġ 2 , Ġ 3 ). to Observable () ĊĊĠĠĠĠĠĠĠ ĠObservable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. sequence Equal ( ob 1 , Ġob 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ Ġb , Ġt Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" on Callback Ġb [$ b ]") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 3 8 _ Schedulers _ Fixed ThreadPool () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġsch Ġ= ĠS chedulers . from ( Executors . new Fixed ThreadPool ( 1 )) ĊĊĠĠĠĠĠĠĠ ĠSingle . just ( Any ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġmap Ġ 1 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 4 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġmap Ġ 1 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( sch ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġmap Ġ 2 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 3 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġmap Ġ 2 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( sch ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġmap Ġ 3 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 2 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġmap Ġ 3 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( sch ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġsubscribe Ġ 4 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġsubscribe Ġ 4 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġprivate Ġval Ġ_ obj Ġ= ĠObject () ĊĊĠĠĠ Ġfun Ġ_ 3 9 _ Object _ wait () Ġ{ ĊĊĠĠĠĠĠĠĠ ĠSingle . just ( Any ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized (_ obj ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġwait Ġsubscribe Ġ 1 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ_ obj . wait () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġwait Ġsubscribe Ġ 1 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 4 0 _ Object _ notify All () Ġ{ ĊĊĠĠĠĠĠĠĠ ĠSingle . just ( Any ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized (_ obj ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġnotify All Ġsubscribe Ġ 1 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ_ obj . notify All () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġnotify All Ġsubscribe Ġ 1 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġ_ re Lock Ġ= ĠRe entrant Lock () ĊĠĠĠ Ġprivate Ġval Ġ_ re Lock Cond Ġ= Ġ_ re Lock . new Condition () ĊĊĊĠĠĠ Ġfun Ġ_ 4 1 _ Re entrant Lock _ await () Ġ{ ĊĊĠĠĠĠĠĠĠ ĠSingle . just ( Any ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe Ġ{ Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġsubscribe Ġ 1 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ_ re Lock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ_ re Lock Cond . await () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġsubscribe Ġ 1 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 4 2 _ Re entrant Lock _ signal All () Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ re Lock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ_ re Lock Cond . signal All () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 4 3 _ Single _ just _ delay () Ġ{ ĊĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġstart ") ĊĊĠĠĠĠĠĠĠ ĠSingle . just ( Any ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġmap 0 ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. delay ( 1 0 , ĠTimeUnit . SECONDS ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Android Schedulers . main Thread ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġmap 1 ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġ_ pub Sub _ signal _ em ul Ġ= ĠPublish Subject . create < Array < Any >>() ĊĊĠĠĠ Ġfun Ġ_ 4 4 _ signal _ em ul () Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ pub Sub _ signal _ em ul ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= Ġit [ 0 ] ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( cmd ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" add " Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġres Ġ= Ġ( it [ 1 ] Ġas ĠInt ) Ġ+ Ġ( it [ 2 ] Ġas ĠInt ) Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġadd Ġres [$ res ]") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= Ġit [ 0 ] ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( cmd ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" sub " Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġres Ġ= Ġ( it [ 1 ] Ġas ĠInt ) Ġ- Ġ( it [ 2 ] Ġas ĠInt ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 3 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d ("[ ${ Thread . currentThread (). name }] Ġsub Ġres [$ res ]") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 4 5 _ signal _ add () Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ pub Sub _ signal _ em ul . onNext ( array Of (" add ", Ġ 1 , Ġ 2 )) ĊĠĠĠĠĠĠĠ Ġ_ pub Sub _ signal _ em ul . onNext ( array Of (" add ", Ġ 3 , Ġ 4 )) ĊĠĠĠĠĠĠĠ Ġ_ pub Sub _ signal _ em ul . onNext ( array Of (" add ", Ġ 5 , Ġ 6 )) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 4 6 _ signal _ sub () Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ pub Sub _ signal _ em ul . onNext ( array Of (" sub ", Ġ 2 , Ġ 1 )) ĊĠĠĠĠĠĠĠ Ġ_ pub Sub _ signal _ em ul . onNext ( array Of (" sub ", Ġ 3 , Ġ 1 )) ĊĠĠĠĠĠĠĠ Ġ_ pub Sub _ signal _ em ul . onNext ( array Of (" sub ", Ġ 4 , Ġ 1 )) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 4 7 _ ByteBuffer _ copy () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbuf Ġ= ĠByteBuffer . wrap (" helloworld ". toByteArray ()) ĊĠĠĠĠĠĠĠ Ġ_ log _ ByteBuffer ( buf , Ġ" buf ") ĊĠĠĠĠĠĠĠ Ġbuf . position ( 2 ) ĊĠĠĠĠĠĠĠ Ġbuf . limit ( 5 ) ĊĠĠĠĠĠĠĠ Ġ_ log _ ByteBuffer ( buf , Ġ" buf ") ĊĠĠĠĠĠĠĠ Ġval Ġbuf 2 Ġ= ĠByteBuffer . allocate ( buf . remaining ()) ĊĠĠĠĠĠĠĠ Ġ_ log _ ByteBuffer ( buf 2 , Ġ" buf 2 ") ĊĠĠĠĠĠĠĠ Ġbuf 2 . put ( buf ) ĊĠĠĠĠĠĠĠ Ġ_ log _ ByteBuffer ( buf 2 , Ġ" buf 2 ") ĊĠĠĠĠĠĠĠ Ġbuf 2 . rewind () ĊĠĠĠĠĠĠĠ Ġ_ log _ ByteBuffer ( buf 2 , Ġ" buf 2 ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġ_ log _ ByteBuffer ( buf : ĠByteBuffer , Ġname : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ ĠLog Ex . d ("${ buf . position ()} :${ buf . limit ()} :${ buf . capacity ()} Ġ$ name [ ${ String ( buf . array ()) }] ") ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġ_ 4 8 _ pipeline _ single _ line () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġps Ġ= ĠPublish Subject . create < Array < Any >>() ĊĊĠĠĠĠĠĠĠ Ġps ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= Ġit [ 0 ] Ġas ĠString ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( cmd ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" cmd 0 " Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" cmd 0 Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" cmd 0 Ġfin ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map ĠarrayOf (" on next ", ĠarrayOf (" cmd 0 ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= Ġit [ 0 ] Ġas ĠString ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( cmd ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" cmd 1 " Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" cmd 1 Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 0 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" cmd 1 Ġfin ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map ĠarrayOf (" on next ", ĠarrayOf (" cmd 1 ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= Ġit [ 0 ] Ġas ĠString ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( cmd ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" on next " Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġon next Array Ġ= Ġit [ 1 ] Ġas ĠArray < Any > ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġin cmd Ġ= Ġon next Array [ 0 ] Ġas ĠString ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" on next Ġin cmd [ ${ in cmd }] ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġps . onNext ( on next Array ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe () ĊĊĊĠĠĠĠĠĠĠ Ġps . onNext ( array Of (" cmd 0 ")) ĊĠĠĠĠĠĠĠ Ġps . onNext ( array Of (" cmd 1 ")) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ_ 4 9 _ pipeline _ double _ line () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġps 0 Ġ= ĠPublish Subject . create < String >() ĊĠĠĠĠĠĠĠ Ġval Ġps 1 Ġ= ĠPublish Subject . create < String >() ĊĊĠĠĠĠĠĠĠ Ġps 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= Ġit Ġas ĠString ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( cmd ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" cmd 0 " Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" cmd 0 Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" cmd 0 Ġfin ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġps 0 . onNext (" cmd 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe () ĊĊĠĠĠĠĠĠĠ Ġps 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. observe On ( Schedulers . io ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcmd Ġ= Ġit Ġas ĠString ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( cmd ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" cmd 1 " Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" cmd 1 Ġstart ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġdelay ( 1 0 _ 0 0 0 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLog Ex . d (" cmd 1 Ġfin ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġps 1 . onNext (" cmd 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe () ĊĊĠĠĠĠĠĠĠ Ġps 0 . onNext (" cmd 0 ") ĊĠĠĠĠĠĠĠ Ġps 1 . onNext (" cmd 1 ") ĊĠĠĠ Ġ} Ċ }