@Test
public void testObservableBasedOpenerAndCloser() {

    Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() {
        @Override
        public Subscription onSubscribe(Observer<? super String> observer) {
            push(observer, "one", 10);
            push(observer, "two", 60);
            push(observer, "three", 110);
            push(observer, "four", 160);
            push(observer, "five", 210);
            complete(observer, 500);
            return Subscriptions.empty();
        }
    });
    Observable<Object> openings = Observable.create(new Observable.OnSubscribeFunc<Object>() {
        @Override
        public Subscription onSubscribe(Observer<Object> observer) {
            push(observer, new Object(), 50);
            push(observer, new Object(), 200);
            complete(observer, 250);
            return Subscriptions.empty();
        }
    });
    Func1<Object, Observable<Object>> closer = new Func1<Object, Observable<Object>>() {
        @Override
        public Observable<Object> call(Object opening) {
            return Observable.create(new Observable.OnSubscribeFunc<Object>() {
                @Override
                public Subscription onSubscribe(Observer<? super Object> observer) {
                    push(observer, new Object(), 100);
                    complete(observer, 101);
                    return Subscriptions.empty();
                }
            });
        }
    };
    Observable<List<String>> buffered = Observable.create(buffer(source, openings, closer));
    buffered.subscribe(observer);
    InOrder inOrder = Mockito.inOrder(observer);
    scheduler.advanceTimeTo(500, TimeUnit.MILLISECONDS);
    inOrder.verify(observer, Mockito.times(1)).onNext(list("two", "three"));
    inOrder.verify(observer, Mockito.times(1)).onNext(list("five"));
    inOrder.verify(observer, Mockito.never()).onNext(Mockito.anyListOf(String.class));
    inOrder.verify(observer, Mockito.never()).onError(Mockito.any(Throwable.class));
    inOrder.verify(observer, Mockito.times(1)).onCompleted();
}