call vs invoke

[PyAntony]

Hi,

I have quick question. Even after reading the "Peeking at the streams" article (https://quarkus.io/blog/mutiny-invoke-and-call/) I am still a bit unsure about the difference between these 2 functions. Basically, can I think of call as a "Fire-and-forget" event? as soon as the Uni returned by call emits an item then just move on with the original computation downstream (Uni is being processed/scheduled in the event loop; whatever happens with that computation we don't care anymore), if it emits a failure then this failure will be propagated downstream. For example:

multi
    . ... // some computation
    .onItem().call(
          () -> Uni.createFrom().item(42).flatmap(__ -> resource.close()) // resource.close() returns Uni<T>
    )
    . ... // downstream computation continues

after item 42 is emitted then original downstream computation continues, it doesn't wait for resource.close() to complete (it is being processed by the vent loop, we just don't care about the response). Am I correct?

[jponge]

Call passes items to an asynchronous operation (hence Uni-returning) but then discards the result of the operation unless it's a failure.

See also https://smallrye.io/smallrye-mutiny/getting-started/transforming-items-async

[PyAntony]

so basically it is just like invoke but it just subscribes to the Uni. Both just run side effects, but invoke runs the consumer (eager) and call executes the lazy value (subscribe). That's all the difference I believe isn't? After testing I find no difference between the 2; if I pass blocking functions to both they just block.

[jponge]

Yes, invoke is meant for consuming synchronously and call is meant for consuming asynchronously.

But it's not auto-magic, if code blocks, it will block even with call (unless you use emitOn / subscribeOn but that's another topic).

[PyAntony]

After some testing I am still confused. Notice the following example:

// runned in main
List<Integer> list = Multi.createFrom().range(0, 10)
    .invoke(i -> System.out.println("Emitting in: " + Thread.currentThread().getName()))
    .call(__ ->
        Uni.createFrom().item(() -> "Any Value")
            .invoke(i -> System.out.println("Emitting (inside 'call') in: " + Thread.currentThread().getName()))
            .onItem().delayIt().by(Duration.ofSeconds(4))
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
    )
    .invoke(i -> System.out.println("Emitting (after 'call') in: " + Thread.currentThread().getName()))
    .collect().asList()
    .await().indefinitely();

Notice the output:

Emitting in: main
Emitting (inside 'call') in: pool-1-thread-1
Emitting (after 'call') in: pool-1-thread-1
Emitting in: pool-1-thread-1
Emitting (inside 'call') in: pool-1-thread-1
Emitting (after 'call') in: pool-1-thread-1
...

So, Multi emits first item in main thread (expected). I follow with call and item is emitted in worker thread since I used runSubscriptionOn (expected). Pipeline waits 4 seconds to continue after call (NOT EXPECTED!). After call I use 'invoke' to see where the main pipeline continues the processing; I see the main thread has been replaced by the worker thread (NOT EXPECTED!)...

Questions:

1. Why am I waiting 4 seconds in call? I send that blocking operation to a worker thread; main thread (where main pipeline is running) should continue immediately without blocking.
2. Why does main pipeline switched from main thread to worker thread? I called runSubscriptionOn in the Uni inside call. I didn't call it in the main pipeline.
3. How can I accomplish my expected operation? Basically I want to send blocking operation to a worker thread, run operation there and forget about it (fire-and-forget, I don't need the result of that computation), but main pipeline should keep running in the thread where the items are being emitted (main thread)

Hope my description is clear, thanks.

[jponge]

That's just the expected behavior, perhaps you have different expectations.

• .await().indefinitely(); is blocking on the main thread until the Multi has completed.
• The inner Uni subscription is indeed switched to a worker pool, and on that same worker pool you will also delay the emission by 4 seconds.