Ошибка на RxJava 2.x печатает странную трассировку стека

Question

Я изучаю RxJava (Абсолютно новый, извините, если этот вопрос слишком базовый), и у меня возникли трудности с механизмом обработки ошибок (я просмотрел документы, но это очень удобно для меня).

Это мой код,

public static void main(String[] args) {
    Observable<String> source = Observable.just("Alpha", "Beta", "Gamma", "Upma", "Idly");
    Observer<String> myObserver = new Observer<String>() {
      @Override
      public void onSubscribe(Disposable d) {
        // do nothing with Disposable, disregard for now
      }
  @Override
  public void onNext(String value) {
    System.out.println("RECEIVED: " + value);
    throw new RuntimeException("I am thrown");
  }

  @Override
  public void onError(Throwable e) {
    System.out.println("I got an error !");
    e.printStackTrace(new PrintStream(System.out));
  }

  @Override
  public void onComplete() {
    System.out.println("Done!");
  }
};
   source.subscribe(myObserver);
  }

И это моя трассировка стека

RECEIVED: Alpha
io.reactivex.exceptions.UndeliverableException: The exception could not be delivered to the consumer because it has already canceled/disposed the flow or the exception has nowhere to go to begin with. Further reading: https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling | java.lang.RuntimeException: I am thrown
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
    at io.reactivex.Observable.subscribe(Observable.java:12275)
    at reactivex.ReactMain.main(ReactMain.java:36)
Caused by: java.lang.RuntimeException: I am thrown
    at reactivex.ReactMain$1.onNext(ReactMain.java:22)
    at reactivex.ReactMain$1.onNext(ReactMain.java:1)
    at io.reactivex.internal.operators.observable.ObservableFromArray$FromArrayDisposable.run(ObservableFromArray.java:108)
    at io.reactivex.internal.operators.observable.ObservableFromArray.subscribeActual(ObservableFromArray.java:37)
    at io.reactivex.Observable.subscribe(Observable.java:12268)
    ... 1 more
Exception in thread "main" io.reactivex.exceptions.UndeliverableException: The exception could not be delivered to the consumer because it has already canceled/disposed the flow or the exception has nowhere to go to begin with. Further reading: https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling | java.lang.RuntimeException: I am thrown
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
    at io.reactivex.Observable.subscribe(Observable.java:12275)
    at reactivex.ReactMain.main(ReactMain.java:36)
Caused by: java.lang.RuntimeException: I am thrown
    at reactivex.ReactMain$1.onNext(ReactMain.java:22)
    at reactivex.ReactMain$1.onNext(ReactMain.java:1)
    at io.reactivex.internal.operators.observable.ObservableFromArray$FromArrayDisposable.run(ObservableFromArray.java:108)
    at io.reactivex.internal.operators.observable.ObservableFromArray.subscribeActual(ObservableFromArray.java:37)
    at io.reactivex.Observable.subscribe(Observable.java:12268)
    ... 1 more
Exception in thread "main" java.lang.NullPointerException: Actually not, but can't throw other exceptions due to RS
    at io.reactivex.Observable.subscribe(Observable.java:12277)
    at reactivex.ReactMain.main(ReactMain.java:36)
Caused by: java.lang.RuntimeException: I am thrown
    at reactivex.ReactMain$1.onNext(ReactMain.java:22)
    at reactivex.ReactMain$1.onNext(ReactMain.java:1)
    at io.reactivex.internal.operators.observable.ObservableFromArray$FromArrayDisposable.run(ObservableFromArray.java:108)
    at io.reactivex.internal.operators.observable.ObservableFromArray.subscribeActual(ObservableFromArray.java:37)
    at io.reactivex.Observable.subscribe(Observable.java:12268)
    ... 1 more

У меня есть два вопроса по этому поводу.

1) Я переопределил onError метод Observer.Почему мой onError() не может перехватить исключения?

2) Даже если onError не удалась (я ожидаю, почему в ответе 1), Почему UndeliverableException выбрасывается только дважды?В идеале, он должен быть брошен 4 раза, так как у меня есть 4 других Observable Strings?

Answer 1

1.

Из onError документации :

Уведомляет наблюдателя о том, что наблюдаемое обнаружило ошибку.

onError не будет вызван, поскольку в исходном источнике не было ошибок.Ошибка была выдана в методе onNext наблюдателя .Если вы хотите проверить onError, вам нужно выдать ошибку в потоке, например:

sourece
    .map( str -> throw new RuntimeException("I am thrown: " + str))
    .subscribe(myObserver);

Выше кода будет вызывать onError вместо onNext.

2. Почему UndeliverableException выдается только дважды?

Я думаю, UndeliverableException выдается только один раз, и все сообщение об ошибке описывает только один сбой.Как только ваш код завершится с ошибкой в ​​методе onNext с «alpha», после этого ничего не произойдет.

Попробуйте запустить код только с одним элементом, например:

Observable<String> source = Observable.just("Alpha");

и посмотрите, получите ли вы то же сообщение об ошибке.Кроме того, вы можете проверить, испускается ли что-нибудь:

Observable<String> source = Observable.just("Alpha", "Beta", "Gamma", "Upma", "Idly")
     .doOnNext(/* put log here to see what is being emitted */);