@Test
public void testWindowViaObservableBoundaryCompletes() {

    PublishSubject<Integer> source = PublishSubject.create();
    PublishSubject<Integer> boundary = PublishSubject.create();
    @SuppressWarnings("unchecked")
    final Observer<Object> o = mock(Observer.class);
    final List<Observer<Object>> values = new ArrayList<Observer<Object>>();
    Observer<Observable<Integer>> wo = new Observer<Observable<Integer>>() {
        @Override
        public void onNext(Observable<Integer> args) {
            @SuppressWarnings("unchecked")
            final Observer<Object> mo = mock(Observer.class);
            values.add(mo);
            args.subscribe(mo);
        }
        @Override
        public void onError(Throwable e) {
            o.onError(e);
        }
        @Override
        public void onCompleted() {
            o.onCompleted();
        }
    };
    source.window(boundary).subscribe(wo);
    int n = 30;
    for (int i = 0; i < n; i++) {
        source.onNext(i);
        if (i % 3 == 2 && i < n - 1) {
            boundary.onNext(i / 3);
        }
    }
    boundary.onCompleted();
    assertEquals(n / 3, values.size());
    int j = 0;
    for (Observer<Object> mo : values) {
        for (int i = 0; i < 3; i++) {
            verify(mo).onNext(j + i);
        }
        verify(mo).onCompleted();
        verify(mo, never()).onError(any(Throwable.class));
        j += 3;
    }
    verify(o).onCompleted();
    verify(o, never()).onError(any(Throwable.class));
}