StoppableObservable < TEntity > getObservable ( final long startIndex , Scheduler scheduler ) { return runAsync ( scheduler , ( rx . Observer < ? super TEntity > observer , Subscription subscription ) -> { long currentStartIndex = startIndex ; while ( ! subscription . isUnsubscribed ( ) ) { PartialResponse < TEntity > response ; try { response = ( currentStartIndex >= 0 ) ? readRange ( currentStartIndex , null ) : readRange ( null , - currentStartIndex ) ; } catch ( IllegalStateException ex ) { // No new data available yet, keep polling continue ; } catch ( IOException | IllegalArgumentException | IllegalAccessException error ) { observer . onError ( error ) ; return ; } response . getElements ( ) . stream ( ) . forEach ( observer :: onNext ) ; if ( response . isEndReached ( ) ) { observer . onCompleted ( ) ; return ; } // Continue polling for more data currentStartIndex = response . getTo ( ) + 1 ; } } ) ; }
Provides an observable stream of elements .