BlockingMultiObserver extends CountDownLatch implements SingleObserver<T>, CompletableObserver, MaybeObserver<T> { public boolean blockingAwait(long timeout, TimeUnit unit) { if (getCount() != 0) { try { BlockingHelper.verifyNonBlocking(); if (!await(timeout, unit)) { dispose(); return false; } } catch (InterruptedException ex) { dispose(); throw ExceptionHelper.wrapOrThrow(ex); } } Throwable ex = error; if (ex != null) { throw ExceptionHelper.wrapOrThrow(ex); } return true; } BlockingMultiObserver(); @Override void onSubscribe(Disposable d); @Override void onSuccess(T value); @Override void onError(Throwable e); @Override void onComplete(); T blockingGet(); T blockingGet(T defaultValue); Throwable blockingGetError(); Throwable blockingGetError(long timeout, TimeUnit unit); boolean blockingAwait(long timeout, TimeUnit unit);  }
@Test public void blockingAwait() { final BlockingMultiObserver<Integer> bmo = new BlockingMultiObserver<Integer>(); Schedulers.single().scheduleDirect(new Runnable() { @Override public void run() { bmo.onSuccess(1); } }, 100, TimeUnit.MILLISECONDS); assertTrue(bmo.blockingAwait(1, TimeUnit.MINUTES)); }