Flux.repeat () не повторяется

Question

Я пытаюсь повторить подписку на Flux следующим образом:

    DirectProcessor<String> stringDirectProcessor = DirectProcessor.create();

    stringDirectProcessor
            .repeat(3)
            .subscribe(item -> System.out.println(item));

    stringDirectProcessor.onNext("one");
    stringDirectProcessor.onNext("two");
    stringDirectProcessor.onNext("three");
    stringDirectProcessor.onComplete();

Я ожидал бы увидеть этот вывод:

one
two
three
one
two
three
one
two
three
one
two
three

Но я получаю только

one
two
three

Однако, если я использую Flux.just() вместо DirectProcessor, я получу ожидаемый результат.

Что не так?

Answer 1

Это ожидаемое поведение DirectProcessor. Я только что прочитал документацию и нашел следующее относительно DirectProcessor:

Как только процессор завершает работу (обычно из-за вызываемой ошибки приемника (Throwable) или complete ()), он позволяет большему количеству подписчиков подписываться, но немедленно передает им сигнал завершения.

Таким образом, поскольку repeat просто повторно подписывается, обработчик onComplete будет вызван для них немедленно. Вы уверены, что вам нужен DirectProcessor?

РЕДАКТИРОВАТЬ: Это поведение также задокументировано здесь

Примечание. Если подписчиков нет, исходящие элементы удаляются и сохраняются только события терминала. Завершенный DirectProcessor будет передавать сигнал терминала последним подписчикам.