WaitMonitor { public void waitFor(long timeout, TimeUnit unit) { try { countDownLatch.await(timeout, unit); synchronized (this) { ReactiveUtil.print("runtime to complete: %s ms", durationToCompleteInMs == null ? "-" : durationToCompleteInMs); } } catch (InterruptedException e) { throw new IllegalStateException(e); } } WaitMonitor(int expectedCompletes);  WaitMonitor(); synchronized void complete(); void waitFor(long timeout, TimeUnit unit); synchronized boolean isComplete();  }
@Test public void testWaitFor() throws Exception { final WaitMonitor monitor = new WaitMonitor(); Observable.just("test") .subscribeOn(Schedulers.computation()) .subscribe(next -> print("next: %s", next), Throwable::printStackTrace, monitor::complete ); monitor.waitFor(100, TimeUnit.MILLISECONDS); assertTrue(monitor.isComplete()); }