Как повторить попытку ошибки в Observable с помощью java-rx? ⇐ JAVA

[Anonymous]

Я пытаюсь изучить Reactive X Java, и мне очень трудно понять, как работает обработка ошибок. То, чего я хочу добиться, просто: Observable генерирует поток данных и иногда может столкнуться с ошибкой. Я хочу передать ошибку подписчикам, но НЕ отписываться от Observable, как это происходит в результате использования onError. В случае возникновения ошибки подписчики должны просто зарегистрировать/обработать ошибку так, как они считают нужным, и сохранить свою подписку. Похоже, что это должна сделать повторная попытка, но я перепробовал все, что мог найти в документации, и не могу заставить работать функцию повторной попытки: в конечном итоге я всегда получаю эту ошибку:
Exception in thread "main" io.reactivex.rxjava3.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/wik ... r-handling | java.lang.Exception: Error: 8

Вот что у меня сейчас, повторные попытки не работают:
public static void main(String[] args) {
ConnectableObservable obs = Observable.create(subscriber -> {
for (int i = 0; i < 10; i++) {
try {
System.out.println("Producer produced data: " + i);
if (i % 2 == 0) {
throw new Exception("Error: " + i);
}
subscriber.onNext(i);
Thread.sleep(500);
} catch (Exception e) {
System.err.println("Producer Error: " + e.getMessage());
subscriber.onError(e);
}
}
}).publish();

obs
.retry(3)
.subscribe(
data -> {
System.out.println("Subscriber 1 received data: " + data);
},
error -> {
System.err.println("Subscriber 1 Error: " + error.getMessage());
},
() -> {
System.out.println("Subscriber 1 Complete");
}
);

obs
.retry(3)
.subscribe(
data -> {
System.out.println("Subscriber 2 received data: " + data);
},
error -> {
System.err.println("Subscriber 2 Error: " + error.getMessage());
},
() -> {
System.out.println("Subscriber 2 Complete");
}
);

obs.connect();
}


Подробнее здесь: https://stackoverflow.com/questions/790 ... th-java-rx