package Ġcom . tl z . rx commons Ċ Ċ import Ġio . reactivex . Observable Ċ import Ġio . reactivex . android . schedul ers . Android Schedulers Ċ import Ġio . reactivex . dis pos ables . Disposable Ċ import Ġio . reactivex . subjects . Publish Subject Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġjava . util . concurrent . TimeUnit . MILLISECONDS Ċ Ċ /** Ċ Ġ* ĠCreated Ġby Ġtom le zen . Ċ Ġ* ĠDate : Ġ 2 0 1 7 / 5 / 2 0 . Ċ Ġ* ĠTime : Ġä¸ĭåįĪ 4 : 1 3 Ċ Ġ*/ Ċ private Ġval Ġsubject Mapper Ġ= ĠConcurrent HashMap < Any , ĠArrayList < Publish Subject < Any >> >() Ċ Ċ fun Ġ< T > Ġon Event ( observable : ĠObservable < T >, Ġon Next : Ġ( T ) Ġ-> ĠUnit , ĠonError : Ġ( Throwable ) Ġ-> ĠUnit ): ĠDis posable Ġ= ĊĠĠĠ Ġobservable . observe On ( Android Schedulers . main Thread ()). subscribe ( onNext , ĠonError ) Ċ Ċ // fun Ġ< T > ĠT . on Event ( observable : ĠObservable < T >, Ġon Next : Ġ( T ) Ġ-> ĠUnit ): ĠDis posable Ġ= Ċ // ĠĠĠ Ġon Event ( observable , Ġon Next , Ġ{ Ġit . printStackTrace () Ġ}) Ċ Ċ fun Ġ< T > Ġon Event ( tag : ĠAny , Ġon Next : Ġ( T ) Ġ-> ĠUnit ): ĠDis posable Ġ= ĊĠĠĠ Ġon Event ( register ( tag ), Ġon Next , Ġ{ Ġit . printStackTrace () Ġ}) Ċ Ċ fun Ġ< T > Ġon Event ( tag : ĠAny , Ġon Next : Ġ( T ) Ġ-> ĠUnit , ĠonError : Ġ( Throwable ) Ġ-> ĠUnit ): ĠDis posable Ġ= ĊĠĠĠ Ġon Event ( register ( tag ), Ġon Next , ĠonError ) Ċ Ċ fun Ġ< T > Ġregister ( tag : ĠAny ): ĠObservable < T > Ġ{ ĊĠ Ġvar Ġsubject List Ġ= Ġsubject Mapper [ tag ] ĊĠ Ġif Ġ( null Ġ== Ġsubject List ) Ġ{ ĊĠĠĠ Ġsubject List Ġ= ĠArrayList () ĊĠĠĠ Ġsubject Mapper [ tag ] Ġ= Ġsubject List ĊĠ Ġ} ĊĠ Ġval Ġsubject Ġ= ĠPublish Subject . create < T >() ĊĠ Ġsubject List . add ( subject Ġas ĠPublish Subject < Any >) ĊĠ Ġreturn Ġsubject Ċ } Ċ Ċ fun Ġunregister ( tag : ĠAny ) Ġ{ ĊĠ Ġif Ġ( subject Mapper . containsKey ( tag )) Ġ{ ĊĠĠĠ Ġsubject Mapper . remove ( tag ) ĊĠ Ġ} Ċ } Ċ Ċ fun Ġunregister ( tag : ĠAny , Ġobservable : ĠObservable <* >) Ġ= ĊĠĠĠ Ġsubject Mapper [ tag ]?. let Ġ{ ĊĠĠĠĠĠ Ġit . remove ( observable ) ĊĠĠĠĠĠ Ġif Ġ( it . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġsubject Mapper . remove ( tag ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊ Ċ fun Ġpost ( content : ĠAny ) Ġ{ ĊĠ Ġpost ( content :: class , Ġcontent ) Ċ } Ċ Ċ fun Ġpost ( tag : ĠAny , Ġcontent : ĠAny ) Ġ{ ĊĠ Ġpost Delay ( tag , Ġcontent , Ġ 0 L ) Ċ } Ċ Ċ fun Ġpost Delay ( content : ĠAny , Ġmillis : ĠLong ): ĠDis posable ? Ġ= ĊĠĠĠ Ġpost Delay ( content , Ġmillis , ĠMIL LI SECONDS ) Ċ Ċ fun Ġpost Delay ( content : ĠAny , Ġdelay : ĠLong , Ġunit : ĠTimeUnit ): ĠDis posable ? Ġ= ĊĠĠĠ Ġpost Delay ( content :: class , Ġcontent , Ġdelay , Ġunit ) Ċ Ċ fun Ġpost Delay ( tag : ĠAny , Ġcontent : ĠAny , Ġmillis : ĠLong ): ĠDis posable ? Ġ= ĊĠĠĠ Ġpost Delay ( tag , Ġcontent , Ġmillis , ĠMIL LI SECONDS ) Ċ Ċ fun Ġpost Delay ( tag : ĠAny , Ġcontent : ĠAny , Ġdelay : ĠLong , Ġunit : ĠTimeUnit ): ĠDis posable ? Ġ= ĊĠĠĠ Ġif Ġ( delay Ġ== Ġ 0 L ) Ġ{ ĊĠĠĠĠĠ Ġsubject Mapper [ tag ]?. filter Ġ{ Ġit . has Observer s () Ġ} ?. forEach Ġ{ Ġit . onNext ( content ) Ġ} ĊĠĠĠĠĠ Ġnull ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠ Ġdelay ( delay , Ġunit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġsubject Mapper [ tag ]?. filter Ġ{ Ġit . has Observer s () Ġ} ?. forEach Ġ{ Ġit . onNext ( content ) Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ