BlockingMultiObserver extends CountDownLatch implements SingleObserver<T>, CompletableObserver, MaybeObserver<T> { public Throwable blockingGetError() { if (getCount() != 0) { try { BlockingHelper.verifyNonBlocking(); await(); } catch (InterruptedException ex) { dispose(); return ex; } } return error; } 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 blockingGetErrorInterrupt() { final BlockingMultiObserver<Integer> bmo = new BlockingMultiObserver<Integer>(); Thread.currentThread().interrupt(); try { assertTrue(bmo.blockingGetError() instanceof InterruptedException); } finally { Thread.interrupted(); } }
@Test public void blockingGetErrorTimeoutInterrupt() { final BlockingMultiObserver<Integer> bmo = new BlockingMultiObserver<Integer>(); Thread.currentThread().interrupt(); try { bmo.blockingGetError(1, TimeUnit.MINUTES); fail("Should have thrown"); } catch (RuntimeException ex) { assertTrue(ex.getCause() instanceof InterruptedException); } finally { Thread.interrupted(); } }