observablebuilder.java

@Test
    public void onComplete() throws Exception {
        TestScheduler testScheduler = new TestScheduler();

        Observable<String> build = ObservableTestBuilder.<String>create(testScheduler).onNext("hans", 500, TimeUnit.MILLISECONDS)
                .onNext("wurst", 500, TimeUnit.MILLISECONDS)
                .onNext("isst", 200, TimeUnit.MILLISECONDS)
                .onNext("gerne", 300, TimeUnit.MILLISECONDS)
                .onNext("fleisch", 1, TimeUnit.SECONDS)
                .onComplete(0, TimeUnit.MILLISECONDS)
                .build();

        TestObserver<List<String>> testSubscriber = build.doOnSubscribe(disposable -> System.out.println("subscribed"))
                .doOnNext(s -> System.out.println("value"))
                .buffer(1001, TimeUnit.MILLISECONDS, testScheduler)
                .test();

        testScheduler.advanceTimeBy(5000, TimeUnit.MILLISECONDS);

        testSubscriber.await()
                .assertNoErrors();

        List<List<String>> values = testSubscriber.values();

        //TODO: TimeCapture
        // TODO: AssertHelper:
        // assert(onNext(t).at(RelativeTime(500)), onNext(t1).at(RelativeTime(500), onComplete().at(any()))

        assertThat(values.get(0), IsIterableContainingInOrder.contains("hans", "wurst"));
        assertThat(values.get(1), IsIterableContainingInOrder.contains("isst", "gerne"));
        assertThat(values.get(2), IsIterableContainingInOrder.contains("fleisch"));
    }