Test.java

package de.cronos.mad.messages.backend;

import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

public class RepeatTest {

    private static class TestSubject {
        public void logMonoValues(Supplier<Mono<String>> monoSupplier) {
            Mono.defer(monoSupplier).repeat().subscribe(System.out::println);
        }
    }

    @Test
    public void repeatTest() {
        DirectProcessor<String> testPublisher = DirectProcessor.create();

        TestSubject testSubject = new TestSubject();
        testSubject.logMonoValues(() -> Mono.from(testPublisher));

        testPublisher.onNext("Hello");
        testPublisher.onComplete();
        testPublisher.onNext("World");
        testPublisher.onComplete();
    }
}