Как гарантировать, что этапы не отбрасывают элементы при появлении первых ошибок восходящего потока (и не нарушают вычис ⇐ JAVA

[Anonymous]

Вопрос1: у меня есть реактивный код, это пример: мне нужно разделить 100 для каждого числа в списке, и я не хочу прерывать вычисления, когда происходит «деление на ноль». Я хочу перейти к следующему номеру, пока не будут использованы все номера в списке. Как можно изменить его для достижения этой цели?

Код: Выделить всё

List list = List.of(1, 2, 3, 4, 5, 0, 7, 8, 9, 10);
CompletionStage future = ReactiveStreams.fromIterable(list)
.map(i -> 100 / i)
.collect(Collectors.summingInt(i -> i))
.run();
future.whenComplete((res, err) -> {
if (err != null) {
System.out.println("error " + err);
} else
System.out.println("res: " + res);
});

Результат консоли:

Код: Выделить всё

error java.lang.ArithmeticException: / by zero

Вопрос 2. Я заменил предыдущий код похожим кодом (ниже), но использую библиотеку Mutiny и хочу добиться того же результата, но происходит то же самое: первое нарушение ошибки вычисление и после цифры 5 вычисление прекращается. Как это исправить? Это код:

Код: Выделить всё

Multi.createFrom().items(1, 2, 3, 4, 5, 0, 7, 8, 9, 10)
.onRequest().invoke(req -> System.out.println("Got a request: " + req))
.onItem().transform(i -> 100/i)
.subscribe().withSubscriber(new Flow.Subscriber() {

private Flow.Subscription subscription;

@Override
public void onSubscribe(Flow.Subscription s) {
this.subscription = s;
s.request(1);
}

@Override
public void onNext(Integer item) {
System.out.println("Got item " + item);
subscription.request(1);
}

@Override
public void onError(Throwable t) {
System.out.println("error " + t);
}

@Override
public void onComplete() {
System.out.println("finish");
}
}
);

Вывод на консоль:

Код: Выделить всё

   Got a request: 1 //1
Got item 100
Got a request: 1 //2
Got item 50
Got a request: 1 //3
Got item 33
Got a request: 1 //4
Got item 25
Got a request: 1 //5
Got item 20   // 100/5 =20
Got a request: 1 //0! and the stop
error java.lang.ArithmeticException: / by zero

Ошибка деления тривиальна, поскольку может быть случайным событием. Например, это может быть вызов n http-ссылок и одна выдает ошибку. я хочу смоделировать ситуацию, которая может произойти... но я бы ИГНОРИРОВАЛ, если это произойдет..

Я попробую использовать
. onFailure().recoverWithItem(0), но после этого он завершается и останавливается вместо продолжения!

Код: Выделить всё

Got a request: 1
Got item 100
Got a request: 1
Got item 50
Got a request: 1
Got item 33
Got a request: 1
Got item 25
Got a request: 1
Got item 20
Got a request: 1
Got item 0
finish

Код:

Код: Выделить всё

Multi.createFrom().items(1, 2, 3, 4, 5, 0, 7, 8, 9, 10)
.onRequest().invoke(req -> System.out.println("Got a request: " + req))
.onItem().transform(i -> 100/i)
.onFailure().recoverWithItem(0)
.subscribe().withSubscriber(new Flow.Subscriber() {

private Flow.Subscription subscription;

@Override
public void onSubscribe(Flow.Subscription s) {
this.subscription = s;
s.request(1);
}

@Override
public void onNext(Integer item) {
System.out.println("Got item " + item);
subscription.request(1);
}

@Override
public void onError(Throwable t) {
System.out.println("error " + t);
}

@Override
public void onComplete() {
System.out.println("finish");
}
}
);

Спасибо
Роби

Подробнее здесь: https://stackoverflow.com/questions/789 ... rrors-arri