@Test public void testSkipAndCountGaplessBuffers() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { observer.onNext("one"); observer.onNext("two"); observer.onNext("three"); observer.onNext("four"); observer.onNext("five"); observer.onCompleted(); return Subscriptions.empty(); } }); Observable<List<String>> buffered = Observable.create(buffer(source, 3, 3)); buffered.subscribe(observer); InOrder inOrder = Mockito.inOrder(observer); inOrder.verify(observer, Mockito.times(1)).onNext(list("one", "two", "three")); inOrder.verify(observer, Mockito.times(1)).onNext(list("four", "five")); inOrder.verify(observer, Mockito.never()).onNext(Mockito.anyListOf(String.class)); inOrder.verify(observer, Mockito.never()).onError(Mockito.any(Throwable.class)); inOrder.verify(observer, Mockito.times(1)).onCompleted(); }
public static <T, TClosing> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Func0<? extends Observable<? extends TClosing>> bufferClosingSelector) { return new OnSubscribeFunc<List<T>>() { @Override public Subscription onSubscribe(Observer<? super List<T>> observer) { NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, OperationBuffer.<T> bufferMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, List<T>, TClosing>(buffers, bufferClosingSelector); return new CompositeSubscription( new ChunkToSubscription(creator), source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator))); } }; }
OperationBuffer extends ChunkedOperation { public static <T, TClosing> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Func0<? extends Observable<? extends TClosing>> bufferClosingSelector) { return new OnSubscribeFunc<List<T>>() { @Override public Subscription onSubscribe(Observer<? super List<T>> observer) { NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, OperationBuffer.<T> bufferMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, List<T>, TClosing>(buffers, bufferClosingSelector); return new CompositeSubscription( new ChunkToSubscription(creator), source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator))); } }; } }
OperationBuffer extends ChunkedOperation { public static <T, TClosing> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Func0<? extends Observable<? extends TClosing>> bufferClosingSelector) { return new OnSubscribeFunc<List<T>>() { @Override public Subscription onSubscribe(Observer<? super List<T>> observer) { NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, OperationBuffer.<T> bufferMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, List<T>, TClosing>(buffers, bufferClosingSelector); return new CompositeSubscription( new ChunkToSubscription(creator), source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator))); } }; }  }
OperationBuffer extends ChunkedOperation { public static <T, TClosing> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Func0<? extends Observable<? extends TClosing>> bufferClosingSelector) { return new OnSubscribeFunc<List<T>>() { @Override public Subscription onSubscribe(Observer<? super List<T>> observer) { NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, OperationBuffer.<T> bufferMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, List<T>, TClosing>(buffers, bufferClosingSelector); return new CompositeSubscription( new ChunkToSubscription(creator), source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator))); } }; }  static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Func0<? extends Observable<? extends TClosing>> bufferClosingSelector); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Observable<? extends TOpening> bufferOpenings, final Func1<? super TOpening, ? extends Observable<? extends TClosing>> bufferClosingSelector); static OnSubscribeFunc<List<T>> buffer(Observable<T> source, int count); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final int count, final int skip); static OnSubscribeFunc<List<T>> buffer(Observable<T> source, long timespan, TimeUnit unit); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final long timespan, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<List<T>> buffer(Observable<T> source, long timespan, TimeUnit unit, int count); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final long timespan, final TimeUnit unit, final int count, final Scheduler scheduler); static OnSubscribeFunc<List<T>> buffer(Observable<T> source, long timespan, long timeshift, TimeUnit unit); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final long timespan, final long timeshift, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<List<T>> bufferWithBoundaryObservable(Observable<? extends T> source, Observable<B> boundary); static OnSubscribeFunc<List<T>> bufferWithBoundaryObservable(Observable<? extends T> source, Observable<B> boundary, int initialCapacity); }
OperationBuffer extends ChunkedOperation { public static <T, TClosing> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Func0<? extends Observable<? extends TClosing>> bufferClosingSelector) { return new OnSubscribeFunc<List<T>>() { @Override public Subscription onSubscribe(Observer<? super List<T>> observer) { NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, OperationBuffer.<T> bufferMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, List<T>, TClosing>(buffers, bufferClosingSelector); return new CompositeSubscription( new ChunkToSubscription(creator), source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator))); } }; }  static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Func0<? extends Observable<? extends TClosing>> bufferClosingSelector); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Observable<? extends TOpening> bufferOpenings, final Func1<? super TOpening, ? extends Observable<? extends TClosing>> bufferClosingSelector); static OnSubscribeFunc<List<T>> buffer(Observable<T> source, int count); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final int count, final int skip); static OnSubscribeFunc<List<T>> buffer(Observable<T> source, long timespan, TimeUnit unit); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final long timespan, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<List<T>> buffer(Observable<T> source, long timespan, TimeUnit unit, int count); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final long timespan, final TimeUnit unit, final int count, final Scheduler scheduler); static OnSubscribeFunc<List<T>> buffer(Observable<T> source, long timespan, long timeshift, TimeUnit unit); static OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final long timespan, final long timeshift, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<List<T>> bufferWithBoundaryObservable(Observable<? extends T> source, Observable<B> boundary); static OnSubscribeFunc<List<T>> bufferWithBoundaryObservable(Observable<? extends T> source, Observable<B> boundary, int initialCapacity);  }