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);  }
@Test public void testComplete() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { observer.onCompleted(); return Subscriptions.empty(); } }); Observable<List<String>> buffered = Observable.create(buffer(source, 3, 3)); buffered.subscribe(observer); Mockito.verify(observer, Mockito.never()).onNext(Mockito.anyListOf(String.class)); Mockito.verify(observer, Mockito.never()).onError(Mockito.any(Throwable.class)); Mockito.verify(observer, Mockito.times(1)).onCompleted(); }
@Test public void testSkipAndCountOverlappingBuffers() { 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"); return Subscriptions.empty(); } }); Observable<List<String>> buffered = Observable.create(buffer(source, 3, 1)); 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("two", "three", "four")); inOrder.verify(observer, Mockito.times(1)).onNext(list("three", "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.never()).onCompleted(); }
@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(); }
@Test public void testSkipAndCountBuffersWithGaps() { 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, 2, 3)); buffered.subscribe(observer); InOrder inOrder = Mockito.inOrder(observer); inOrder.verify(observer, Mockito.times(1)).onNext(list("one", "two")); 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(); }
@Test public void testTimedAndCount() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { push(observer, "one", 10); push(observer, "two", 90); push(observer, "three", 110); push(observer, "four", 190); push(observer, "five", 210); complete(observer, 250); return Subscriptions.empty(); } }); Observable<List<String>> buffered = Observable.create(buffer(source, 100, TimeUnit.MILLISECONDS, 2, scheduler)); buffered.subscribe(observer); InOrder inOrder = Mockito.inOrder(observer); scheduler.advanceTimeTo(100, TimeUnit.MILLISECONDS); inOrder.verify(observer, Mockito.times(1)).onNext(list("one", "two")); scheduler.advanceTimeTo(200, TimeUnit.MILLISECONDS); inOrder.verify(observer, Mockito.times(1)).onNext(list("three", "four")); scheduler.advanceTimeTo(300, TimeUnit.MILLISECONDS); inOrder.verify(observer, Mockito.times(1)).onNext(list("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(); }
@Test public void testTimed() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { push(observer, "one", 98); push(observer, "two", 99); push(observer, "three", 100); push(observer, "four", 101); push(observer, "five", 102); complete(observer, 150); return Subscriptions.empty(); } }); Observable<List<String>> buffered = Observable.create(buffer(source, 100, TimeUnit.MILLISECONDS, scheduler)); buffered.subscribe(observer); InOrder inOrder = Mockito.inOrder(observer); scheduler.advanceTimeTo(101, TimeUnit.MILLISECONDS); inOrder.verify(observer, Mockito.times(1)).onNext(list("one", "two", "three")); scheduler.advanceTimeTo(201, TimeUnit.MILLISECONDS); 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(); }
@Test public void testObservableBasedOpenerAndCloser() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { push(observer, "one", 10); push(observer, "two", 60); push(observer, "three", 110); push(observer, "four", 160); push(observer, "five", 210); complete(observer, 500); return Subscriptions.empty(); } }); Observable<Object> openings = Observable.create(new Observable.OnSubscribeFunc<Object>() { @Override public Subscription onSubscribe(Observer<Object> observer) { push(observer, new Object(), 50); push(observer, new Object(), 200); complete(observer, 250); return Subscriptions.empty(); } }); Func1<Object, Observable<Object>> closer = new Func1<Object, Observable<Object>>() { @Override public Observable<Object> call(Object opening) { return Observable.create(new Observable.OnSubscribeFunc<Object>() { @Override public Subscription onSubscribe(Observer<? super Object> observer) { push(observer, new Object(), 100); complete(observer, 101); return Subscriptions.empty(); } }); } }; Observable<List<String>> buffered = Observable.create(buffer(source, openings, closer)); buffered.subscribe(observer); InOrder inOrder = Mockito.inOrder(observer); scheduler.advanceTimeTo(500, TimeUnit.MILLISECONDS); inOrder.verify(observer, Mockito.times(1)).onNext(list("two", "three")); inOrder.verify(observer, Mockito.times(1)).onNext(list("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(); }
@Test public void testObservableBasedCloser() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { push(observer, "one", 10); push(observer, "two", 60); push(observer, "three", 110); push(observer, "four", 160); push(observer, "five", 210); complete(observer, 250); return Subscriptions.empty(); } }); Func0<Observable<Object>> closer = new Func0<Observable<Object>>() { @Override public Observable<Object> call() { return Observable.create(new Observable.OnSubscribeFunc<Object>() { @Override public Subscription onSubscribe(Observer<? super Object> observer) { push(observer, new Object(), 100); complete(observer, 101); return Subscriptions.empty(); } }); } }; Observable<List<String>> buffered = Observable.create(buffer(source, closer)); buffered.subscribe(observer); InOrder inOrder = Mockito.inOrder(observer); scheduler.advanceTimeTo(500, TimeUnit.MILLISECONDS); inOrder.verify(observer, Mockito.times(1)).onNext(list("one", "two")); inOrder.verify(observer, Mockito.times(1)).onNext(list("three", "four")); inOrder.verify(observer, Mockito.times(1)).onNext(list("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(); }
@Test public void testLongTimeAction() throws InterruptedException { final CountDownLatch latch = new CountDownLatch(1); LongTimeAction action = new LongTimeAction(latch); Observable.from(1).buffer(10, TimeUnit.MILLISECONDS, 10) .subscribe(action); latch.await(); assertFalse(action.fail); }
@Test public void testBufferStopsWhenUnsubscribed1() { Observable<Integer> source = Observable.never(); Observer<List<Integer>> o = mock(Observer.class); Subscription s = source.buffer(100, 200, TimeUnit.MILLISECONDS, scheduler).subscribe(o); InOrder inOrder = Mockito.inOrder(o); scheduler.advanceTimeBy(1001, TimeUnit.MILLISECONDS); inOrder.verify(o, times(5)).onNext(Arrays.<Integer> asList()); s.unsubscribe(); scheduler.advanceTimeBy(999, TimeUnit.MILLISECONDS); inOrder.verifyNoMoreInteractions(); }
@Test public void bufferWithBONormal1() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> boundary = PublishSubject.create(); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = Mockito.inOrder(o); source.buffer(boundary).subscribe(o); source.onNext(1); source.onNext(2); source.onNext(3); boundary.onNext(1); inOrder.verify(o, times(1)).onNext(Arrays.asList(1, 2, 3)); source.onNext(4); source.onNext(5); boundary.onNext(2); inOrder.verify(o, times(1)).onNext(Arrays.asList(4, 5)); source.onNext(6); boundary.onCompleted(); inOrder.verify(o, times(1)).onNext(Arrays.asList(6)); inOrder.verify(o).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void bufferWithBOEmptyLastViaBoundary() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> boundary = PublishSubject.create(); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = Mockito.inOrder(o); source.buffer(boundary).subscribe(o); boundary.onCompleted(); inOrder.verify(o, times(1)).onNext(Arrays.asList()); inOrder.verify(o).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void bufferWithBOEmptyLastViaSource() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> boundary = PublishSubject.create(); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = Mockito.inOrder(o); source.buffer(boundary).subscribe(o); source.onCompleted(); inOrder.verify(o, times(1)).onNext(Arrays.asList()); inOrder.verify(o).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void bufferWithBOEmptyLastViaBoth() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> boundary = PublishSubject.create(); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = Mockito.inOrder(o); source.buffer(boundary).subscribe(o); source.onCompleted(); boundary.onCompleted(); inOrder.verify(o, times(1)).onNext(Arrays.asList()); inOrder.verify(o).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void bufferWithBOSourceThrows() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> boundary = PublishSubject.create(); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); source.buffer(boundary).subscribe(o); source.onNext(1); source.onError(new OperationReduceTest.CustomException()); verify(o).onError(any(OperationReduceTest.CustomException.class)); verify(o, never()).onCompleted(); verify(o, never()).onNext(any()); }
@Test public void bufferWithBOBoundaryThrows() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> boundary = PublishSubject.create(); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); source.buffer(boundary).subscribe(o); source.onNext(1); boundary.onError(new OperationReduceTest.CustomException()); verify(o).onError(any(OperationReduceTest.CustomException.class)); verify(o, never()).onCompleted(); verify(o, never()).onNext(any()); }