BlockingObservable { public T singleOrDefault(T defaultValue) { return blockForSingle(o.map(UtilityFunctions.<T>identity()).singleOrDefault(defaultValue)); } 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 testSingleOrDefaultUnsubscribe() throws InterruptedException { final CountDownLatch unsubscribe = new CountDownLatch(1); Observable<Integer> o = Observable.create(new OnSubscribe<Integer>() { @Override public void call(Subscriber<? super Integer> subscriber) { subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { unsubscribe.countDown(); } })); subscriber.onNext(1); subscriber.onNext(2); } }).subscribeOn(Schedulers.newThread()); try { o.toBlocking().singleOrDefault(-1); fail("Expected IllegalArgumentException because there are 2 elements"); } catch (IllegalArgumentException e) { } assertTrue("Timeout means `unsubscribe` is not called", unsubscribe.await(30, TimeUnit.SECONDS)); }