@Test public void testTimedAndCount() { final List<String> list = new ArrayList<String>(); final List<List<String>> lists = new ArrayList<List<String>>(); 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<Observable<String>> windowed = Observable.create(window(source, 100, TimeUnit.MILLISECONDS, 2, scheduler)); windowed.subscribe(observeWindow(list, lists)); scheduler.advanceTimeTo(100, TimeUnit.MILLISECONDS); assertEquals(1, lists.size()); assertEquals(lists.get(0), list("one", "two")); scheduler.advanceTimeTo(200, TimeUnit.MILLISECONDS); assertEquals(2, lists.size()); assertEquals(lists.get(1), list("three", "four")); scheduler.advanceTimeTo(300, TimeUnit.MILLISECONDS); assertEquals(3, lists.size()); assertEquals(lists.get(2), list("five")); }
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);  }