package Ġcom . m ire go . tri kot . streams . reactive . processors Ċ Ċ import Ġcom . m ire go . tri kot . foundation . concurrent . Atomic Reference Ċ import Ġcom . m ire go . tri kot . streams . reactive . Streams Processor Exception Ċ import Ġorg . reactiv est ream s . Publisher Ċ import Ġorg . reactiv est ream s . Subscriber Ċ Ċ type alias ĠScan Processor Block < T > Ġ= Ġ( acc : ĠT , Ġcurrent : ĠT ) Ġ-> ĠT Ċ Ċ class ĠScan Processor < T >( parent Publisher : ĠPublisher < T >, Ġprivate Ġval Ġinitial Value : ĠT ?, Ġprivate Ġval Ġblock : ĠScan Processor Block < T >) Ġ: ĊĠĠĠ ĠAbstract Processor < T , ĠT >( parent Publisher ) Ġ{ ĊĊĠĠĠ Ġoverride Ġfun Ġcreate Subscription ( subscriber : ĠSubscriber < in ĠT >): ĠProcessor Subscription < T , ĠT > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠScan Processor Subscription ( subscriber , Ġinitial Value , Ġblock ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠScan Processor Subscription < T >( ĊĠĠĠĠĠĠĠ Ġsubscriber : ĠSubscriber < in ĠT >, ĊĠĠĠĠĠĠĠ Ġinitial Value : ĠT ?, ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġblock : ĠScan Processor Block < T > ĊĠĠĠ Ġ) Ġ: ĠProcessor Subscription < T , ĠT >( subscriber ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġatomic Value Ġ= ĠAtomic Reference < T ? >( initial Value ) ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( t : ĠT , Ġsubscriber : ĠSubscriber < in ĠT >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġv Ġ= Ġatomic Value . value ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( v Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġatomic Value . set OrThrow ( null , Ġt ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . onNext ( t ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġblock ( v , Ġt ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( exception : ĠStreams Processor Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonError ( exception ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġatomic Value . set OrThrow ( v , Ġresult ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . onNext ( result ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ