  @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();
  }