BlockingObservable { public void forEach(final Action1<? super T> onNext) { final CountDownLatch latch = new CountDownLatch(1); final AtomicReference<Throwable> exceptionFromOnError = new AtomicReference<Throwable>(); Subscription subscription = o.subscribe(new Subscriber<T>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { exceptionFromOnError.set(e); latch.countDown(); } @Override public void onNext(T args) { onNext.call(args); } }); try { latch.await(); } catch (InterruptedException e) { subscription.unsubscribe(); Thread.currentThread().interrupt(); throw new RuntimeException("Interrupted while waiting for subscription to complete.", e); } if (exceptionFromOnError.get() != null) { if (exceptionFromOnError.get() instanceof RuntimeException) { throw (RuntimeException) exceptionFromOnError.get(); } else { throw new RuntimeException(exceptionFromOnError.get()); } } } private  BlockingObservable(Observable<? extends T> o); static BlockingObservable<T> from(final Observable<? extends T> o); void forEach(final Action1<? super T> onNext); Iterator<T> getIterator(); T first(); T first(Func1<? super T, Boolean> predicate); T firstOrDefault(T defaultValue); T firstOrDefault(T defaultValue, Func1<? super T, Boolean> predicate); T last(); T last(final Func1<? super T, Boolean> predicate); T lastOrDefault(T defaultValue); T lastOrDefault(T defaultValue, Func1<? super T, Boolean> predicate); Iterable<T> mostRecent(T initialValue); Iterable<T> next(); Iterable<T> latest(); T single(); T single(Func1<? super T, Boolean> predicate); T singleOrDefault(T defaultValue); T singleOrDefault(T defaultValue, Func1<? super T, Boolean> predicate); Future<T> toFuture(); Iterable<T> toIterable();  }
@Test public void testUnsubscribeFromForEachWhenInterrupted() throws InterruptedException { new InterruptionTests().assertUnsubscribeIsInvoked("forEach()", new Action1<BlockingObservable<Void>>() { @Override public void call(final BlockingObservable<Void> o) { o.forEach(new Action1<Void>() { @Override public void call(final Void aVoid) { } }); } }); }