Testing subscription with repeat()

I want to test a function that uses the repeat() operator on a deferred mono and subscribes to the result. In the test, I use the TestPublisher to simulate new values from the mono.

In a very simplified form, it looks like this:

package de.cronos.mad.messages.backend;

import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.publisher.TestPublisher;

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() {
        TestPublisher<String> testPublisher = TestPublisher.create();

        TestSubject testSubject = new TestSubject();
        testSubject.logMonoValues(testPublisher::mono);

        testPublisher.emit("Hello");
        testPublisher.emit("World");
    }
}

The execution hangs after logging "Hello" to stdout. I think I understand why this happens: The emit(…) call happens from the main thread and "drives" the subscription from there.

What I do not know is how to modify this test so that it completes, i.e. does not hang?

Notes:

1. The .repeat() operator when used without arguments means to repeat indefinitely. It initiates a new subscription immediately once the original one completes successfully. Two ways to terminate it are:

   1. Limit the count of iterations, e.g. as .repeat(10), as suggested by vins;

   2. Abort the sequence by sending an error signal, e.g. with Mono.error(Throwable) or just by throwing an exception.

2. Calling testPublisher::mono declares that this instance of TestPublisher adheres to a Mono contract, i.e. only sends a single value.

   Thus neither calling .emit() twice, nor calling it with two arguments will help. Extra values will be ignored.

3. The javadoc for TestPublisher says,

   > TestPublisher are generally hot, [...] replaying the first termination signal to subsequent subscribers.

   It means that late subscribers immediately receive the termination signal. The subscriptions created by repeat() operator immediately receive the replayed termination signal, which in turn causes repeat() to resubscribe, in a tight loop.

I think that a possible solution is to use Mono.defer() or Mono.fromSupplier() to create a new value for each subscription created by .repeat(). E.g.:

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Supplier;

import reactor.core.publisher.Mono;

public class So63117029 {

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

  public static void main(String[] args) {
    LinkedBlockingQueue<String> data = new LinkedBlockingQueue<>(List.of("Hello", "World"));

    TestSubject testSubject = new TestSubject();
    testSubject.logMonoValues(() -> Mono.fromSupplier(data::remove));
  }
}

The method Queue.remove() throws a NoSuchElementException when there is no more data in the queue.

The TestPublisher emit method accepts array of values. Once emitted, it closes the source. So you can not emit one by one. Instead pass all the values like this.

testPublisher.emit("Hello", "world");

// from TestPublisher emit
public final TestPublisher<T> emit(T... values) {
    Objects.requireNonNull(values, "values array is null, please cast to T if null T required");
    Object[] var2 = values;
    int var3 = values.length;

    for(int var4 = 0; var4 < var3; ++var4) {
        T t = var2[var4];
        this.next(t);
    }

    return this.complete();
}

Regarding the hanging behavior, the repeat here seems to cause the issue as it indefinitely tries to reconnect the source Mono again and again.

Mono.defer(monoSupplier).repeat().subscribe(System.out::println);

Just change it like this to understand the behavior.

Mono.defer(monoSupplier).repeat(10).subscribe(System.out::println);