Моно асинхронная обработка исключений

Question

Я просто пытаюсь понять, как работает обработка исключений в библиотеке реакторов.

Рассмотрим следующий пример:

public class FluxTest {

@Test
public void testIt() throws InterruptedException {
    Scheduler single = Schedulers.single();

    CountDownLatch latch = new CountDownLatch(1);

    Mono.just("hey")
            .subscribeOn(single)
            .doOnError(e -> System.out.println("ERROR1: " + e.getMessage()))
            .doOnTerminate((r, e) -> {
                if (e != null) System.out.println("ERROR3: " + e.getMessage());
                latch.countDown();
            })
            .subscribe(
                    it -> { throw new RuntimeException("Error for " + it); },
                    ex -> { System.out.println("ERROR2: " + ex.getMessage());}
                    );

    latch.await();
}

}

На самом деле, я не вижу ни одного из выполняемых блоков кода обработки ошибок. Тест завершается без какого-либо сообщения. Также я пытался удалить doOnError, doOnTerminate Процессоры без удачи.

Answer 1

И это правильное поведение в вашем случае. Вы создаете моно из единственной строки "эй", в которой нет ошибок. Если вы попытаетесь отладить, вы можете увидеть, что doOnTerminate метод вызывается с e = null параметр и согласно документации вызывается в любом случае success или же error,

Чтобы проверить некоторую обработку ошибок, вы можете сделать следующее:

public class FluxTest {
    @Test
    public void testIt() throws InterruptedException {
        Scheduler single = Schedulers.single();

        CountDownLatch latch = new CountDownLatch(1);

        Mono.just("hey")
                .map(this::mapWithException)
                .subscribeOn(single)
                .doOnError(e -> System.out.println("ERROR1: " + e.getMessage()))
                .doOnTerminate((r, e) -> {
                    if (e != null) System.out.println("ERROR3: " + e.getMessage());
                    latch.countDown();
                })
                .subscribe(
                        it -> { throw new RuntimeException("Error for " + it); },
                        ex -> { System.out.println("ERROR2: " + ex.getMessage());}
                        );

        latch.await();
    }

    private String mapWithException(String s) {
        throw new RuntimeException();
    }
}

После запуска теста с кодом выше вы должны три строки в вашей консоли

ERROR1: null
ERROR3: null
ERROR2: null

Итак, первый обратный вызов onError когда моно не удалось, второй будет onTerminate потому что моно завершается с ошибкой, а третий errorConsumer от subscribe метод.