Как продолжить последовательность fluxMap после исключения ⇐ JAVA

[Anonymous]

Я новичок в проекте «Реактор» и столкнулся с проблемой. Вот пример кода:

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

// Message has a type of SomeClass
flux.flatMap(message -> someService.someOperationReturningMono(...)
.flatMap(i -> anotherService.operationWichCanProducerException(message.getProperty()))
.flatMap(j -> anotherService.operationWichCanProducerException(message.getProperty())))
.then()
.onErrorContinue(...)
.subscribe(...);

Проблема в том, что первый вызов может вызвать исключение, и из-за этого второй вызов не произойдет, потому что, как я знаю, после первого исключения в цепочке оригинальный Publisher завершает работу, но для меня это важно чтобы гарантировать второй вызов метода службы. Более того, в будущем таких вызовов будет больше, потому что фактические параметры всегда разные. Есть ли способ гарантировать, что если во время вызова метода службы возникнет какое-либо исключение, методы службы из следующих плоских карт все равно будут вызываться?
РЕДАКТИРОВАТЬ 1:
Я хочу добавить некоторый контекст к моей проблеме: метод из службы А предполагает отправку сообщений Kafka (асинхронно, поэтому тип возвращаемого значения их методов — Mono) с другим типом (веб-сообщение, сообщение EMAIL и т. д.). ) по конкретной теме (одинаково для каждого сообщения). Вот временное решение, которое, похоже, работает с методом seriveA, просто печатая строку в консоли (еще не проверял это с помощью Kafka)

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

Mono mono = Mono.just(new FakeDbRecord("Propety"));

mono.flatMap(message -> serviceC.someRandomLogic()
.flatMap(i -> serviceA.methodA(message.getSomeProperty(), Types.WEB).then(Mono.just(message)))
.onErrorResume(e -> {
System.out.println("BOOOOOOOOOOM!!!!! " + e.getMessage());

return Mono.just(message);
})
.flatMap(j -> serviceA.methodA(message.getSomeProperty(), Types.EMAIL).then(Mono.just(message)))
.onErrorResume(e -> {
System.out.println("BOOOOOOOOOOM!!!!! " + e.getMessage());

return Mono.just(message);
})).subscribe();

В реальном коде я получаю запись из БД (сообщение), затем эта запись обрабатывается какой-то случайной логикой (serviceC.someRandomLogic()) и если все в порядке - мы отправляют уведомления в kafka
Действительно ли это хорошее решение моей проблемы? Или, может быть, есть способы получше?

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