@Test public void testNonOverlappingWindows() { Observable<String> subject = Observable.from("one", "two", "three", "four", "five"); Observable<Observable<String>> windowed = Observable.create(window(subject, 3)); List<List<String>> windows = toLists(windowed); assertEquals(2, windows.size()); assertEquals(list("one", "two", "three"), windows.get(0)); assertEquals(list("four", "five"), windows.get(1)); }
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);  }