Uni#runSubscriptionOn effect downstream

[gwenneg]

Hi! I have a question regarding the following code snippet:

Uni.createFrom().voidItem()
        .invoke(() -> System.out.println("Thread #1: " + Thread.currentThread().getName()))
        .onItem().call(() ->
                        Uni.createFrom().item(() -> {
                            System.out.println("Thread #2: " + Thread.currentThread().getName());
                            return (Void) null;
                        }).runSubscriptionOn(Infrastructure.getDefaultExecutor())
        )
        .invoke(() -> System.out.println("Thread #3: " + Thread.currentThread().getName()))

After reading the Uni#runSubscriptionOn Javadoc, I thought the runSubscriptionOn above would only affect the inner Uni and wouldn't affect the downstream invoke, but it does.

I expected that kind of output:

Thread #1: vert.x-eventloop-thread-47
Thread #2: executor-thread-0
Thread #1: vert.x-eventloop-thread-47

while I actually get this:

Thread #1: vert.x-eventloop-thread-47
Thread #2: executor-thread-0
Thread #3: executor-thread-0

Is there a way, in a single stream, to switch from the event loop to a worker thread and then switch back to the event loop?

Edit:

I should probably explain the reason behind my question. 😃

I need to execute RestAssured blocking requests between reactive operations from test code. I thought I could run the RestAssured requests from a worker thread using runSubscriptionOn to avoid blocking the event loop and keep the preceding and following reactive calls execution on the event loop. In the snippet above, println #1 and #3 would be reactive calls and println #2 would be a RestAssured request.

Maybe there's a better approach than using runSubscriptionOn (which might not work at all) to mix reactive and RestAssured calls in test code.

[jponge]

Hi @gwenneg,

It's not too surprising that calls get trampolined on the same thread.

Is there a way, in a single stream, to switch from the event loop to a worker thread and then switch back to the event loop?

If it's a pure Vert.x question you can use vertx.executeBlocking(...) to offset some operation to a worker thread, then process the result back on the event-loop.

In more advanced cases you can also use a Vert.x Context and call runOnContext(...) to execute an action on its event-loop.

Remember that Mutiny is orthogonal of the threading models, so you can:

1. move subscription to different threads as you did, and/or
2. move emissions to different threads with emitOn.

Hope it helps.

[gwenneg]

Thank you for your answer @jponge.

Vert.x is executing the code under the hood indeed, but I was hoping there would be a pure Mutiny solution for my question.

When I first read the Imperative to Reactive section of the doc, I understood (because of the isolate word) that it was possible to mix operations executed on both the event loop and worker threads in a single reactive stream with Mutiny:

Let’s imagine you have blocking code (e.g., connecting to a database using JDBC, reading a file from the file system…​), and you want to integrate that into your reactive pipelines while avoiding blocking. You would need to isolate such blocking parts of your code and run these parts on worker threads.

[jponge]

I tend to think that in your case you should look at emitOn, since the blocking operations are while processing items.

Also note that there are helpers at https://smallrye.io/smallrye-mutiny-vertx-bindings/2.13.0/apidocs/io/smallrye/mutiny/vertx/MutinyHelper.html to help you dispatch back to a Vert.x event-loop "executor" (but you need a vertx reference which you likely can get anyway).

[gwenneg]

A quick test of emitOn combined with MutinyHelper seems to produce the result I'm looking for:

Uni.createFrom().voidItem()
        .invoke(() -> System.out.println("Thread #1: " + Thread.currentThread().getName()))
        .call(() ->
                        Uni.createFrom().item(() -> {
                            System.out.println("Thread #2: " + Thread.currentThread().getName());
                            return (Void) null;
                        }).runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
        )
        .emitOn(MutinyHelper.executor(vertx))
        .call(() -> {
            return Uni.createFrom().voidItem()
                    .invoke(() -> System.out.println("Thread #3: " + Thread.currentThread().getName()));
        })

Thread #1: vert.x-eventloop-thread-47
Thread #2: executor-thread-0
Thread #3: vert.x-eventloop-thread-0

I'll test that with the real code (which contains Hibernate Reactive queries) tomorrow. Thanks for the pointers!

[jponge]

Note that you might want to use the Vert.x context and not the Vert.x object if you want to execute back on the same event-loop thread.

[gwenneg]

That's even better, thanks!