@Test public void completeCompletesInner() { Observable<Message> messages = Observable.just(new Message("Bob", "Hello")); final AtomicInteger seen = new AtomicInteger(); WindowIfChanged.create(messages, userSelector) .switchMap( new Function<GroupedObservable<String, Message>, Observable<Notification<String>>>() { @Override public Observable<Notification<String>> apply( GroupedObservable<String, Message> group) { final int count = seen.incrementAndGet(); return group.map(new Function<Message, String>() { @Override public String apply(Message message) throws Exception { return count + " " + message; } }).materialize(); } }) .test() .assertValues( Notification.createOnNext("1 Bob Hello"), Notification.<String>createOnComplete()) .assertComplete(); }
public static <T, K> Observable<GroupedObservable<K, T>> create(Observable<T> upstream, Function<? super T, ? extends K> keySelector) { return new WindowIfChangedObservable<>(upstream, keySelector); }
WindowIfChanged { public static <T, K> Observable<GroupedObservable<K, T>> create(Observable<T> upstream, Function<? super T, ? extends K> keySelector) { return new WindowIfChangedObservable<>(upstream, keySelector); } }
WindowIfChanged { public static <T, K> Observable<GroupedObservable<K, T>> create(Observable<T> upstream, Function<? super T, ? extends K> keySelector) { return new WindowIfChangedObservable<>(upstream, keySelector); } private  WindowIfChanged(); }
WindowIfChanged { public static <T, K> Observable<GroupedObservable<K, T>> create(Observable<T> upstream, Function<? super T, ? extends K> keySelector) { return new WindowIfChangedObservable<>(upstream, keySelector); } private  WindowIfChanged(); static Observable<GroupedObservable<K, T>> create(Observable<T> upstream,
      Function<? super T, ? extends K> keySelector); }
WindowIfChanged { public static <T, K> Observable<GroupedObservable<K, T>> create(Observable<T> upstream, Function<? super T, ? extends K> keySelector) { return new WindowIfChangedObservable<>(upstream, keySelector); } private  WindowIfChanged(); static Observable<GroupedObservable<K, T>> create(Observable<T> upstream,
      Function<? super T, ? extends K> keySelector);  }