public static < T > Observable < T > toColdObservable ( final RecursiveFutureProvider < T > futureProvider ) { return Observable . create ( new Observable . OnSubscribe < T > ( ) { @ Override public void call ( final Subscriber < ? super T > subscriber ) { recursiveChain ( subscriber , futureProvider . createStopCriteria ( ) ) ; } private void recursiveChain ( final Subscriber < ? super T > subscriber , final Predicate < T > stopCriteria ) { futureProvider . provide ( ) . consume ( result -> { if ( result . isSuccess ( ) ) { final T value = result . getValue ( ) ; subscriber . onNext ( value ) ; if ( stopCriteria . apply ( value ) ) { subscriber . onCompleted ( ) ; } else { recursiveChain ( subscriber , stopCriteria ) ; } } else { subscriber . onError ( result . getError ( ) ) ; } } ) ; } } ) ; }
creates new cold observable given future provider on each subscribe will consume the provided future and repeat until stop criteria will exists each result will be emitted to the stream