LambdaObserver extends AtomicReference<Disposable> implements Observer<T>, Disposable, LambdaConsumerIntrospection { @Override public boolean isDisposed() { return get() == DisposableHelper.DISPOSED; } LambdaObserver(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
            Action onComplete,
            Consumer<? super Disposable> onSubscribe); @Override void onSubscribe(Disposable s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void dispose(); @Override boolean isDisposed(); @Override boolean hasCustomOnError();  }
@Test public void onSubscribeThrows() { final List<Object> received = new ArrayList<Object>(); LambdaObserver<Object> o = new LambdaObserver<Object>(new Consumer<Object>() { @Override public void accept(Object v) throws Exception { received.add(v); } }, new Consumer<Throwable>() { @Override public void accept(Throwable e) throws Exception { received.add(e); } }, new Action() { @Override public void run() throws Exception { received.add(100); } }, new Consumer<Disposable>() { @Override public void accept(Disposable s) throws Exception { throw new TestException(); } }); assertFalse(o.isDisposed()); Observable.just(1).subscribe(o); assertTrue(received.toString(), received.get(0) instanceof TestException); assertEquals(received.toString(), 1, received.size()); assertTrue(o.isDisposed()); }
@Test public void onNextThrows() { final List<Object> received = new ArrayList<Object>(); LambdaObserver<Object> o = new LambdaObserver<Object>(new Consumer<Object>() { @Override public void accept(Object v) throws Exception { throw new TestException(); } }, new Consumer<Throwable>() { @Override public void accept(Throwable e) throws Exception { received.add(e); } }, new Action() { @Override public void run() throws Exception { received.add(100); } }, new Consumer<Disposable>() { @Override public void accept(Disposable s) throws Exception { } }); assertFalse(o.isDisposed()); Observable.just(1).subscribe(o); assertTrue(received.toString(), received.get(0) instanceof TestException); assertEquals(received.toString(), 1, received.size()); assertTrue(o.isDisposed()); }
@Test public void onErrorThrows() { List<Throwable> errors = TestHelper.trackPluginErrors(); try { final List<Object> received = new ArrayList<Object>(); LambdaObserver<Object> o = new LambdaObserver<Object>(new Consumer<Object>() { @Override public void accept(Object v) throws Exception { received.add(v); } }, new Consumer<Throwable>() { @Override public void accept(Throwable e) throws Exception { throw new TestException("Inner"); } }, new Action() { @Override public void run() throws Exception { received.add(100); } }, new Consumer<Disposable>() { @Override public void accept(Disposable s) throws Exception { } }); assertFalse(o.isDisposed()); Observable.<Integer>error(new TestException("Outer")).subscribe(o); assertTrue(received.toString(), received.isEmpty()); assertTrue(o.isDisposed()); TestHelper.assertError(errors, 0, CompositeException.class); List<Throwable> ce = TestHelper.compositeList(errors.get(0)); TestHelper.assertError(ce, 0, TestException.class, "Outer"); TestHelper.assertError(ce, 1, TestException.class, "Inner"); } finally { RxJavaPlugins.reset(); } }
@Test public void onCompleteThrows() { List<Throwable> errors = TestHelper.trackPluginErrors(); try { final List<Object> received = new ArrayList<Object>(); LambdaObserver<Object> o = new LambdaObserver<Object>(new Consumer<Object>() { @Override public void accept(Object v) throws Exception { received.add(v); } }, new Consumer<Throwable>() { @Override public void accept(Throwable e) throws Exception { received.add(e); } }, new Action() { @Override public void run() throws Exception { throw new TestException(); } }, new Consumer<Disposable>() { @Override public void accept(Disposable s) throws Exception { } }); assertFalse(o.isDisposed()); Observable.<Integer>empty().subscribe(o); assertTrue(received.toString(), received.isEmpty()); assertTrue(o.isDisposed()); TestHelper.assertUndeliverable(errors, 0, TestException.class); } finally { RxJavaPlugins.reset(); } }