WebFlux: как работает метод takeUntilOther ()?

Question

Может кто-нибудь объяснить, как работает метод takeUntilOther()? Я попытался запустить следующий код, но он ничего не показывает на моей консоли.

     Mono.just(10)
                .subscribe();

        Flux.range(1, 5)
                .takeUntilOther(Mono.just(10))
                .subscribe(System.out::println);

Я не понимаю, почему.

Answer 1

Кирилл,

Я бы предложил вам сослаться на соответствующую часть документации реактора проекта .

takeUntilOther(Publisher<?> other) Передача значений из этого потока до тех пор, пока данный издатель не выдаст.

То есть вы будете получать значения из исходного потока до тех пор, пока Publisher<?> other не начнет генерировать события. В вашем случае у вас есть горячий издатель just(), который немедленно прерывает исходный поток (вызывая метод cancel()).

Я приведу еще один пример. Посмотрите на следующий фрагмент кода:

Flux.range(1, 5) // produces elements from 1 to 5
        .delayElements(Duration.ofSeconds(1)) // delays emission of each element from above for 1 second
        .takeUntilOther(Mono
                .just(10) // hot publisher. emits one element

                // delays '10' for 3 seconds. meaning that it will only 
                // appears in the original Flux in 3 seconds
                .delayElement(Duration.ofSeconds(3)) 
        )
        .subscribe(System.out::print);

Вывод:

12