@Test public void testWindowViaObservableBoundaryThrows() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> boundary = PublishSubject.create(); @SuppressWarnings("unchecked") final Observer<Object> o = mock(Observer.class); final List<Observer<Object>> values = new ArrayList<Observer<Object>>(); Observer<Observable<Integer>> wo = new Observer<Observable<Integer>>() { @Override public void onNext(Observable<Integer> args) { @SuppressWarnings("unchecked") final Observer<Object> mo = mock(Observer.class); values.add(mo); args.subscribe(mo); } @Override public void onError(Throwable e) { o.onError(e); } @Override public void onCompleted() { o.onCompleted(); } }; source.window(boundary).subscribe(wo); source.onNext(0); source.onNext(1); source.onNext(2); boundary.onError(new OperationReduceTest.CustomException()); assertEquals(1, values.size()); Observer<Object> mo = values.get(0); verify(mo).onNext(0); verify(mo).onNext(1); verify(mo).onNext(2); verify(mo).onError(any(OperationReduceTest.CustomException.class)); verify(o, never()).onCompleted(); verify(o).onError(any(OperationReduceTest.CustomException.class)); }
public static <T, TClosing> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Func0<? extends Observable<? extends TClosing>> windowClosingSelector) { return new OnSubscribeFunc<Observable<T>>() { @Override public Subscription onSubscribe(final Observer<? super Observable<T>> observer) { NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, OperationWindow.<T> windowMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, Observable<T>, TClosing>(windows, windowClosingSelector); return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator)); } }; }
OperationWindow extends ChunkedOperation { public static <T, TClosing> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Func0<? extends Observable<? extends TClosing>> windowClosingSelector) { return new OnSubscribeFunc<Observable<T>>() { @Override public Subscription onSubscribe(final Observer<? super Observable<T>> observer) { NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, OperationWindow.<T> windowMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, Observable<T>, TClosing>(windows, windowClosingSelector); return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator)); } }; } }
OperationWindow extends ChunkedOperation { public static <T, TClosing> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Func0<? extends Observable<? extends TClosing>> windowClosingSelector) { return new OnSubscribeFunc<Observable<T>>() { @Override public Subscription onSubscribe(final Observer<? super Observable<T>> observer) { NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, OperationWindow.<T> windowMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, Observable<T>, TClosing>(windows, windowClosingSelector); return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator)); } }; }  }
OperationWindow extends ChunkedOperation { public static <T, TClosing> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Func0<? extends Observable<? extends TClosing>> windowClosingSelector) { return new OnSubscribeFunc<Observable<T>>() { @Override public Subscription onSubscribe(final Observer<? super Observable<T>> observer) { NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, OperationWindow.<T> windowMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, Observable<T>, TClosing>(windows, windowClosingSelector); return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator)); } }; }  static Func0<Window<T>> windowMaker(); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Func0<? extends Observable<? extends TClosing>> windowClosingSelector); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Observable<? extends TOpening> windowOpenings, final Func1<? super TOpening, ? extends Observable<? extends TClosing>> windowClosingSelector); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, int count); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final int count, final int skip); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit, int count); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final TimeUnit unit, final int count, final Scheduler scheduler); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, long timeshift, TimeUnit unit); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final long timeshift, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, Observable<U> boundary); }
OperationWindow extends ChunkedOperation { public static <T, TClosing> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Func0<? extends Observable<? extends TClosing>> windowClosingSelector) { return new OnSubscribeFunc<Observable<T>>() { @Override public Subscription onSubscribe(final Observer<? super Observable<T>> observer) { NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, OperationWindow.<T> windowMaker()); ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, Observable<T>, TClosing>(windows, windowClosingSelector); return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator)); } }; }  static Func0<Window<T>> windowMaker(); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Func0<? extends Observable<? extends TClosing>> windowClosingSelector); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Observable<? extends TOpening> windowOpenings, final Func1<? super TOpening, ? extends Observable<? extends TClosing>> windowClosingSelector); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, int count); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final int count, final int skip); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit, int count); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final TimeUnit unit, final int count, final Scheduler scheduler); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, long timeshift, TimeUnit unit); static OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final long timeshift, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, Observable<U> boundary);  }