AssertSubscriber working with generator() based Multi stream

[victpork]

Hello, I am writing test cases for the Multi stream and trying to get the AssertSubscriber working. Here are my test code:

@Test
void multiTest() {
    Multi<String> multi = Multi.createFrom().generator(() -> 0, (i, em) -> {
        if (i > 5) {
            em.complete();
        } else {
            em.emit(List.of("i=" + i, "i=" + i));
        }

        return i+1;
    }).onItem().<String>disjoint().onItem().transform(i -> i);
    AssertSubscriber<String> subscriber = multi.subscribe().withSubscriber(AssertSubscriber.create(10));
    subscriber.assertCompleted().assertItems("i=0","i=0","i=1","i=1","i=2","i=2","i=3","i=3");
}

but it never reach completion and fails:

Expected a completion event, but didn't received it.
The subscriber received item events:
<i=0,i=0,i=1,i=1,i=2,i=2,i=3,i=3,i=4,i=4>
java.lang.AssertionError: 
Expected a completion event, but didn't received it.
The subscriber received item events:
<i=0,i=0,i=1,i=1,i=2,i=2,i=3,i=3,i=4,i=4>
	at io.smallrye.mutiny.helpers.test.AssertionHelper.fail(AssertionHelper.java:146)
	at io.smallrye.mutiny.helpers.test.AssertionHelper.shouldHaveCompleted(AssertionHelper.java:21)
	at io.smallrye.mutiny.helpers.test.AssertSubscriber.assertCompleted(AssertSubscriber.java:196)
	at io.equalink.EntityTest.multiTest(EntityTest.java:137)
	at java.base/java.lang.reflect.Method.invoke(Method.java:565)
	at io.quarkus.test.junit.QuarkusTestExtension.runExtensionMethod(QuarkusTestExtension.java:998)
	at io.quarkus.test.junit.QuarkusTestExtension.interceptTestMethod(QuarkusTestExtension.java:846)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1604)

Wrote another case without the AssertSubscriber, it works as expected:

@Test
void multiTestWithoutAssertComplete() {
    Multi<String> multi = Multi.createFrom().generator(() -> 0, (i, em) -> {
        if (i > 5) {
            em.complete();
        } else {
            em.emit(List.of("i=" + i, "i=" + i));
        }

        return i+1;
    }).onItem().disjoint();
    CompletableFuture<Void> syncFlag = new CompletableFuture<>();
    List<String> results = new ArrayList<>();
    multi.subscribe().with(results::add, () -> syncFlag.complete(null));
    syncFlag.join();
    assertThat(results, hasItems("i=0","i=0","i=1","i=1","i=2","i=2","i=3","i=3"));
}

Am I writing it wrong? Or does it not work with generator() ?
Thanks

[jponge]

You make a request of (up to) 10 items, but you generate pairs in lists, then disjoint picks those items to "flatten" the list, so at the generator level you haven't reached i=5 yet. If you change the initial request to 20, you will receive a completion.

Remember that demand applies to the last operator in the pipeline first, so requesting 10 in your example applies to the disjoint operator, not to the generator.