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;
            }
        });
    }
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.