package Ġcom . m ire go . tri kot . streams . reactive . processors Ċ Ċ import Ġcom . m ire go . tri kot . streams . c ancellable . C ancellable Manager Ċ import Ġcom . m ire go . tri kot . streams . reactive . Publish ers Ċ import Ġcom . m ire go . tri kot . streams . reactive . Streams Processor Exception Ċ import Ġcom . m ire go . tri kot . streams . reactive . subscribe Ċ import Ġcom . m ire go . tri kot . streams . reactive . take While Ċ import Ġkotlin . test . Test Ċ import Ġkotlin . test . assertEquals Ċ import Ġkotlin . test . assert Fails With Ċ import Ġkotlin . test . assertTrue Ċ Ċ class ĠTake While Processor Tests Ġ{ ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Take While () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= ĠPublish ers . behavior Subject ( 0 ) ĊĠĠĠĠĠĠĠ Ġval Ġreceived Results Ġ= Ġmutable ListOf < Int >() ĊĠĠĠĠĠĠĠ Ġvar Ġcompleted Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ Ġpublisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. take While Ġ{ Ġit Ġ!= Ġ 3 Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠC ancellable Manager (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Completed Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompleted Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ 1 ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ 2 ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ 3 ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ 4 ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( 0 , Ġ 1 , Ġ 2 ), Ġreceived Results ) ĊĠĠĠĠĠĠĠ ĠassertTrue ( completed ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Completion Behavior When Predicate IsTrue And Source Publisher Is Completed () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= ĠPublish ers . just ( true ) ĊĠĠĠĠĠĠĠ Ġval Ġreceived Results Ġ= Ġmutable ListOf < Boolean >() ĊĠĠĠĠĠĠĠ Ġvar Ġcompleted Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ Ġpublisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. take While Ġ{ Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠC ancellable Manager (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Completed Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompleted Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( listOf ( true ), Ġreceived Results ) ĊĠĠĠĠĠĠĠ ĠassertTrue ( completed ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Completion Behavior When Predicate IsFalse And Source Publisher Is Completed () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= ĠPublish ers . just ( false ) ĊĠĠĠĠĠĠĠ Ġval Ġreceived Results Ġ= Ġmutable ListOf < Boolean >() ĊĠĠĠĠĠĠĠ Ġvar Ġcompleted Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ Ġpublisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. take While Ġ{ Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠC ancellable Manager (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Completed Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompleted Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( emptyList (), Ġreceived Results ) ĊĠĠĠĠĠĠĠ ĠassertTrue ( completed ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Re connection When Predicate Is Always True () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= ĠPublish ers . publish Subject < String >() ĊĠĠĠĠĠĠĠ Ġval Ġreceived Results Ġ= Ġmutable ListOf < String >() ĊĊĠĠĠĠĠĠĠ Ġval Ġtake While Publisher Ġ= Ġpublisher . take While Ġ{ Ġit Ġ!= Ġ" d " Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġcancell able Manager 1 Ġ= ĠC ancellable Manager () ĊĠĠĠĠĠĠĠ Ġval Ġcancell able Manager 2 Ġ= ĠC ancellable Manager () ĊĊĠĠĠĠĠĠĠ Ġtake While Publisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( c ancellable Manager 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" a " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" b " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" c " ĊĊĠĠĠĠĠĠĠ Ġcancell able Manager 1 . cancel () ĊĊĠĠĠĠĠĠĠ Ġtake While Publisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( c ancellable Manager 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" a " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" b " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" c " ĊĊĠĠĠĠĠĠĠ Ġcancell able Manager 2 . cancel () ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( listOf (" a ", Ġ" b ", Ġ" c ", Ġ" a ", Ġ" b ", Ġ" c "), Ġreceived Results ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Re connection When Predicate IsFalse () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= ĠPublish ers . publish Subject < String >() ĊĠĠĠĠĠĠĠ Ġval Ġreceived Results Ġ= Ġmutable ListOf < String >() ĊĊĠĠĠĠĠĠĠ Ġval Ġtake While Publisher Ġ= Ġpublisher . take While Ġ{ Ġit Ġ!= Ġ" c " Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġcancell able Manager 1 Ġ= ĠC ancellable Manager () ĊĠĠĠĠĠĠĠ Ġval Ġcancell able Manager 2 Ġ= ĠC ancellable Manager () ĊĊĠĠĠĠĠĠĠ Ġtake While Publisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( c ancellable Manager 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" a " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" b " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" c " ĊĊĠĠĠĠĠĠĠ Ġtake While Publisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( c ancellable Manager 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" a " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" b " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" c " ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( listOf (" a ", Ġ" b ", Ġ" a ", Ġ" b "), Ġreceived Results ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Should Complete If Predicate IsFalse () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= ĠPublish ers . publish Subject < String >() ĊĠĠĠĠĠĠĠ Ġval Ġreceived Results Ġ= Ġmutable ListOf < String >() ĊĊĠĠĠĠĠĠĠ Ġval Ġtake While Publisher Ġ= Ġpublisher . take While Ġ{ Ġit Ġ!= Ġ" b " Ġ} ĊĊĠĠĠĠĠĠĠ Ġvar Ġfirst Completion Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġvar Ġsecond Completion Ġ= Ġfalse ĊĊĠĠĠĠĠĠĠ Ġtake While Publisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠC ancellable Manager (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Completed Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfirst Completion Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" a " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" b " ĊĊĠĠĠĠĠĠĠ Ġtake While Publisher ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠC ancellable Manager (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Results . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Completed Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsecond Completion Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" a " ĊĠĠĠĠĠĠĠ Ġpublisher . value Ġ= Ġ" b " ĊĊĠĠĠĠĠĠĠ ĠassertTrue ( first Completion ) ĊĠĠĠĠĠĠĠ ĠassertTrue ( second Completion ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Mapping Streams Processor Exception () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= ĠPublish ers . behavior Subject (" a ") ĊĠĠĠĠĠĠĠ Ġval Ġexpected Exception Ġ= ĠStreams Processor Exception () ĊĠĠĠĠĠĠĠ Ġvar Ġreceived Exception : ĠStreams Processor Exception ? Ġ= Ġnull ĊĊĠĠĠĠĠĠĠ Ġpublisher . take While Ġ{ Ġthrow Ġexpected Exception Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠC ancellable Manager (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Next Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Exception Ġ= Ġit Ġas ĠStreams Processor Exception ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( expected Exception , Ġreceived Exception ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Mapping Any Exception () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpublisher Ġ= ĠPublish ers . behavior Subject (" a ") ĊĊĠĠĠĠĠĠĠ Ġassert Fails With ( IllegalStateException :: class ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpublisher . take While Ġ{ Ġthrow ĠIllegalStateException () Ġ}. subscribe ( C ancellable Manager ()) Ġ{} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ