core/src/main/java/net/typedrest/AbstractStreamEndpoint.java
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;
            }
        });
    }
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;
            }
        });
    }
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.

@param startIndex The index of the first element to return in the stream.
Use negative values to start counting from the end of the stream.
@param scheduler The scheduler used to run the background thread.
@return An observable stream of elements.
Provides an observable stream of elements .