Reactor-core: Mono.fromRunnable(...) never emits where Mono.fromCallable(...) does

[Jamalarm]

We encountered what seems like a bug (could be wrong!)

We have found that this code works (all events are asynchonously passed to processEvent):

replayEventsStream
            .flatMap(event -> Mono.fromCallable(() -> {
                pipelines.get(event.getQueueId()).processEvent(event);
                return event;
            }).subscribeOn(Schedulers.elastic()), concurrency)
            .doOnComplete(completionHook::countDown)
            .doOnError((e) -> completionHook.countDown())
            .subscribe();

while this code will complete immediately with no events processed via processEvent

replayEventsStream
            .flatMap(event -> Mono.fromRunnable(() -> {
                pipelines.get(event.getQueueId()).processEvent(event);
            }).subscribeOn(Schedulers.elastic()), concurrency)
            .doOnComplete(completionHook::countDown)
            .doOnError((e) -> completionHook.countDown())
            .subscribe();

The only difference is fromCallable() vs fromRunnable(). Any ideas?

[akarnokd]

Barebone unit test:

@Test
    public void test() {
        int c[] = { 0 };
        Flux.range(1, 1000)
        .flatMap(v -> Mono.fromRunnable(() -> { c[0]++; }))
        .ignoreElements()
        .block();

        Assert.assertEquals(1000, c[0]);
    }

[smaldini]

Hey thanks for the report, please use hide() in front of fromRunnable() as a workaround is indeed a bug that will be fixed in a sec.

[yu199195]

ues hide ？

[yu199195]

now i use Mono.fromRunnable() ERROR reactor.core.publisher.Operators - Operator called default onErrorDropped by webflux-http

[AhmedGoudaa]

Still can't do maping using V3.2.11.RELEASE

Mono<String> fromCallable =
 Mono.fromRunnable(() -> {
      System.out.println("fromCallable ");
    }).map(anyVal -> {
      System.out.println(" in mapping Mapping ");
      return "any";
    });

    fromCallable_.subscribe();

prints "fromCallable" only .
any help ?