@Test public void splits() { Observable<Message> messages = Observable.just( new Message("Bob", "Hello"), new Message("Bob", "World"), new Message("Alice", "Hey"), new Message("Bob", "What's"), new Message("Bob", "Up?"), new Message("Eve", "Hey") ); 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.createOnNext("1 Bob World"), Notification.<String>createOnComplete(), Notification.createOnNext("2 Alice Hey"), Notification.<String>createOnComplete(), Notification.createOnNext("3 Bob What's"), Notification.createOnNext("3 Bob Up?"), Notification.<String>createOnComplete(), Notification.createOnNext("4 Eve Hey"), Notification.<String>createOnComplete()); }
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);  }