Multi await feature for intentional blocking

[danielkec]

There was a request for blocking feature for Multi and Single

Multi.interval(20, TimeUnit.MILLISECONDS, Executors.newSingleThreadScheduledExecutor())
        .limit(5)
        .forEach(sum::addAndGet)
        .await();

Multi.interval(20, TimeUnit.MILLISECONDS, Executors.newSingleThreadScheduledExecutor())
        .limit(5)
        .reduce(Long::sum)
        .await();

I guess this needs some broader discussion, it kind of feels like introducing an antipattern. But the truth is other impls has similar feature and users will be expecting it.

Signed-off-by: Daniel Kec daniel.kec@oracle.com

[akarnokd]

I have mixed feelings about blocking operators and since Project Loom is not fixed yet, it's unclear how we can play nice with it as well.

[romain-grecourt]

Can you rebase against master ? you're missing one of the checks from the new infra.

[danielkec]

@tomas-langer had proposed Single implementing CompletionStage which is seems to be quite nicer, so Multi.forEach can returnSingle<Void>, await methods are part of single on one place only and its possible to return Single on the apis where completables are expected

[danielkec]

Can I ask @romain-grecourt and @tomas-langer for review? @Verdent is blocked by this with #1646 Question is if Single implementing CompletionStage is preferable or not, Imho I like it as it makes all apis returning Single more comfortable.

[romain-grecourt]

@danielkec Yeah, Let's go ahead with Single implementing CompletionStage, we can leverage it in our APIs where we return a CompletionStage.

[tomas-langer]

LGTM.
We may want to revisit the completion stage single, but this is a good first change

[danielkec]

Awaits are now preserved when chaining CompletionStage methods:

        testMulti()
                .forEach(sum::addAndGet)
                .whenComplete((aVoid, throwable) -> {})
                .thenRun(() -> {})
                .thenAccept(aVoid -> {})
                .await();

[tomas-langer]

LGTM

[olotenko]

This seems like a massive breaking change just to get a wrapper for a timed CompletableFuture.get.

onCancel implementation is incompatible in its behaviour with superclass. Multi.forEach can miss cancellation indefinitely. Dangling Subscribers are possible.

[danielkec]

@olotenko pls elaborate, is it order of?:

                    subscription.request(Long.MAX_VALUE);
                    single.onCancel(subscription::cancel);

[olotenko]

Yes, but don't you see the whole design of the change is rickety?

A plain static method for two flavours of await would serve the purpose. I really don't see the value of rewriting CompletableFuture and changing how all the reactive pipelines everywhere behave.