FlowableThrowingTest.java 文件源码

@Test public void bodyThrowingInOnCompleteDeliveredToPlugin() {
  server.enqueue(new MockResponse());

  final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
  RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
    @Override public void accept(Throwable throwable) throws Exception {
      if (!throwableRef.compareAndSet(null, throwable)) {
        throw Exceptions.propagate(throwable);
      }
    }
  });

  RecordingSubscriber<String> subscriber = subscriberRule.create();
  final RuntimeException e = new RuntimeException();
  service.body().subscribe(new ForwardingSubscriber<String>(subscriber) {
    @Override public void onComplete() {
      throw e;
    }
  });

  subscriber.assertAnyValue();
  assertThat(throwableRef.get()).isSameAs(e);

}