/* Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 1 8 ĠE man uel ĠM ach ado Ġda ĠSil va Ġ< eman uel . m ch @ gmail . com > Ċ Ġ* Ċ Ġ* ĠPermission Ġis Ġhereby Ġgranted , Ġfree Ġof Ġcharge , Ġto Ġany Ġperson Ġobtaining Ġa Ġcopy Ċ Ġ* Ġof Ġthis Ġsoftware Ġand Ġassociated Ġdocumentation Ġfiles Ġ( the Ġ" Software "), Ġto Ġdeal Ċ Ġ* Ġin Ġthe ĠSoftware Ġwithout Ġrestriction , Ġincluding Ġwithout Ġlimitation Ġthe Ġrights Ċ Ġ* Ġto Ġuse , Ġcopy , Ġmodify , Ġmerge , Ġpublish , Ġdistribute , Ġsublicense , Ġand / or Ġsell Ċ Ġ* Ġcopies Ġof Ġthe ĠSoftware , Ġand Ġto Ġpermit Ġpersons Ġto Ġwhom Ġthe ĠSoftware Ġis Ċ Ġ* Ġfurnished Ġto Ġdo Ġso , Ġsubject Ġto Ġthe Ġfollowing Ġconditions : Ċ Ġ* Ċ Ġ* ĠThe Ġabove Ġcopyright Ġnotice Ġand Ġthis Ġpermission Ġnotice Ġshall Ġbe Ġincluded Ġin Ġall Ċ Ġ* Ġcopies Ġor Ġsubstantial Ġportions Ġof Ġthe ĠSoftware . Ċ Ġ* Ċ Ġ* ĠTHE ĠSOFTWARE ĠIS ĠPROVIDED Ġ" AS ĠIS ", ĠWITHOUT ĠWARRANTY ĠOF ĠANY ĠKIND , ĠEXPRESS ĠOR Ċ Ġ* ĠIMPLIED , ĠINCLUDING ĠBUT ĠNOT ĠLIMITED ĠTO ĠTHE ĠWARRANTIES ĠOF ĠMERCHANTABILITY , Ċ Ġ* ĠFITNESS ĠFOR ĠA ĠPARTICULAR ĠPURPOSE ĠAND ĠNONINFRINGEMENT . ĠIN ĠNO ĠEVENT ĠSHALL ĠTHE Ċ Ġ* ĠAUTHORS ĠOR ĠCOPYRIGHT ĠHOLDERS ĠBE ĠLIABLE ĠFOR ĠANY ĠCLAIM , ĠDAMAGES ĠOR ĠOTHER Ċ Ġ* ĠLIABILITY , ĠWHETHER ĠIN ĠAN ĠACTION ĠOF ĠCONTRACT , ĠTORT ĠOR ĠOTHERWISE , ĠARISING ĠFROM , Ċ Ġ* ĠOUT ĠOF ĠOR ĠIN ĠCONNECTION ĠWITH ĠTHE ĠSOFTWARE ĠOR ĠTHE ĠUSE ĠOR ĠOTHER ĠDEALINGS ĠIN ĠTHE Ċ Ġ* ĠSOFTWARE . Ċ Ġ*/ Ċ Ċ package Ġbill . reak t ive Ċ Ċ import Ġjava . util . concurrent . Count DownLatch Ċ Ċ // FIX ME : ĠMake Ġthis Ġthread - safe Ċ internal Ġclass ĠBlock ing Last Subscriber < T > Ġ: ĠSubscriber < T > Ġ{ ĊĠĠĠ Ġprivate Ġvar Ġlatest Value : ĠT ? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġvar Ġis Finished Ġ= Ġfalse ĊĠĠĠ Ġprivate Ġval Ġcount DownLatch Ġ= ĠCount DownLatch ( 1 ) ĊĊĠĠĠ Ġoverride Ġfun Ġon Next ( element : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlatest Value Ġ= Ġelement ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ ĊĠĠĠĠĠĠĠ Ġis Finished Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġcount DownLatch . count Down () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Cancel () Ġ{ ĊĠĠĠĠĠĠĠ Ġis Finished Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġcount DownLatch . count Down () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonError ( error : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠ Ġerror . printStackTrace () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsubscribe To ( publisher : ĠPublisher < T >): ĠT Ġ{ ĊĠĠĠĠĠĠĠ Ġpublisher . subscribe ( this ) ĊĊĠĠĠĠĠĠĠ Ġif Ġ( is Finished . not ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcount DownLatch . await () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġlatest Value Ġ?: Ġthrow ĠIllegalStateException (" Publisher Ġfailed Ġto Ġpublish Ġany Ġvalues ") ĊĠĠĠ Ġ} Ċ } Ċ Ċ internal Ġclass ĠEmpty Subscriber < T > Ġ: ĠSubscriber < T > Ġ{ ĊĠĠĠ Ġoverride Ġfun Ġon Next ( element : ĠT ) Ġ{} ĊĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{} ĊĠĠĠ Ġoverride Ġfun Ġon Cancel () Ġ{} ĊĠĠĠ Ġoverride Ġfun ĠonError ( error : ĠThrowable ) Ġ{} Ċ } Ċ Ċ class ĠTest Subscriber < T > Ġinternal Ġconstructor ( publisher : ĠPublisher < T >) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġsubscription : ĠSubscription ĊĠĠĠ Ġprivate Ġvar Ġdid Complete : ĠBoolean Ġ= Ġfalse ĊĠĠĠ Ġprivate Ġval Ġemitted Values Ġ= Ġmutable ListOf < T >() ĊĊĠĠĠ Ġval Ġemitted Errors Ġ= Ġmutable Set Of < Throwable >() ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġsubscription Ġ= Ġpublisher ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Complete Ġ{ Ġthis . did Complete Ġ= Ġtrue Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġthis . emit ted Values Ġ+= Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do OnError Ġ{ Ġthis . emit ted Errors Ġ+= Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcancel (): ĠTest Subscriber < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġsubscription . cancel () ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġassert Complete (): ĠTest Subscriber < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( did Complete ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġthrow ĠAssertionError (" Expected Ġsubscription Ġto Ġhave Ġreceived Ġan Ġ` onComplete ` Ġsignal ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġassert Not Complete (): ĠTest Subscriber < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( did Complete ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġjava . lang . AssertionError (" Expected Ġsubscription Ġto ĠNOT Ġhave Ġreceived Ġan Ġ` onComplete ` Ġsignal ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġassert Emit ted Values ( var arg Ġelements : ĠT ): ĠTest Subscriber < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( elements . toList () Ġ!= Ġemitted Values ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠAssertionError (" Values Ġare Ġwrong , Ġexpected Ġ[$ elements ] Ġbut Ġwas Ġ[$ emit ted Values ]") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġassert No Values Emit ted (): ĠTest Subscriber < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( emit ted Values . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġjava . lang . AssertionError (" Expected Ġno Ġelements Ġno Ġhave Ġbeen Ġemitted , Ġbut Ġwas Ġ[$ emit ted Values ]") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġassert No Errors (): ĠTest Subscriber < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( emit ted Errors . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠAssertionError (" Expected Ġno Ġerrors , Ġbut Ġthere Ġwere Ġ${ emit ted Errors . size } Ġerrors ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinline Ġfun Ġ< re ified ĠE Ġ: ĠThrowable > Ġassert Error (): ĠTest Subscriber < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġemitted Errors . isEmpty () Ġ-> Ġthrow ĠAssertionError (" Expected Ġ[ ${ E :: class }] Ġbut Ġno Ġerrors Ġhave Ġbeen Ġemitted ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġemitted Errors . size Ġ> Ġ 1 Ġ-> Ġthrow ĠAssertionError (" Expected Ġsingle Ġerror Ġ[ ${ E :: class }] Ġbut Ġmultiple Ġerrors Ġhave Ġbeen Ġemitted : Ġ[$ emit ted Errors ]") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġemitted Errors . first () Ġ! is ĠE Ġ-> Ġthrow ĠAssertionError (" Expected Ġsingle Ġerror Ġ[ ${ E :: class }] Ġbut Ġonly Ġ[ ${ emit ted Errors . first ()} ] Ġhas Ġbeen Ġemitted ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} Ċ } Ċ