Help needed with Mutiny reactive programming

[franck102]

Hi all,

I am getting my feet wet with Mutiny and quite frankly struggling a bit despite reading every tutorial & posts I could find.
From my Quarkus app I need to download a file from a Stripe API, store that file as a local temp file, then store the file is AWS S3.

I have a reactive REST client that seems to be working (code below), however I am not quite sure how I should set up the Uni chain. This attempt fails with a BlockingNotAllowedException; I read that the @Blocking annotation allows Mutiny to automatically decide whether to switch to a worker thread, but that doesn't work in that context?

The exception (stack trace below) happens even if I uncomment the "runSubscriptionOn" line... so I am kind of stuck :(

 client.download()
                // .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
                .subscribe().with(in -> storeInvoice(invoice, in),
                    e -> {
                        _log.error("Could not download invoice", e);
                        throw new RegisterException(500, ErrorCode.InternalError, e);
                    });
...
 @Blocking
    private void storeInvoice(Invoice invoice, InputStream in)
    {
        ...
    }

The exception stack trace:

14:36:35 [] ERROR [io.qu.mu.ru.MutinyInfrastructure] (vert.x-eventloop-thread-5) Mutiny had to drop the following exception: org.jboss.resteasy.reactive.common.core.BlockingNotAllowedException: Attempting a blocking read on io thread
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream$VertxBlockingInput.readBlocking(VertxClientInputStream.java:199)
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream.readIntoBuffer(VertxClientInputStream.java:81)
	at org.jboss.resteasy.reactive.client.handlers.VertxClientInputStream.read(VertxClientInputStream.java:62)
	at java.base/java.io.InputStream.transferTo(InputStream.java:782)
	at com.easypass.register.service.InvoiceService.storeInvoice(InvoiceService.java:217)

The REST Client:

DownloadClient client = RestClientBuilder.newBuilder()
                .baseUri(new URI(downloadURL)).followRedirects(true).build(DownloadClient.class);

...

@Path("")
public interface DownloadClient
{
    @Path("")
    @GET
    @Consumes(MediaType.WILDCARD)
    Uni<InputStream> download();
}

Any pointers welcome!

[quarkus-bot[bot]]

/cc @cescoffier (mutiny), @jponge (mutiny)

[jponge]

Something that could work would be like this sketch, where the subscription would be handled by Quarkus instead of your own code:

@Path("/abc")
public Uni<Something> endpoint() {
  return downloadClient.download()
    .onItem().transformToUni(data -> s3client.upload(data));
}

Now if you do need to store to a file between the HTTP client request and the S3 upload, I'm afraid blocking I/O can't be avoided so reactive won't bring much value as you will always need to offset that particular step to a worker thread (runSubscriptionOn can indeed be used for that).

[franck102]

Fundamentally what I don't understand is why

.runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
                .subscribe().with(in -> in.read(...)

generates the exception?

Is it the case that that VertxClientInputStream cannot be used in any thread but the I/O thread that handles the HTTP request (this is what I meant by Uni<InputStream> being a non-starter)?
Or did I get the "runSubscriptionOn" pattern wrong, since the exception is thrown from the event-loop thread (vert.x-eventloop-thread-5) ?

[jponge]

It's because the subscription starts from a worker, triggers a HTTP client request but that request response will be processed on a Vert.x event loop, so the next steps in your code do blocking I/O on the inputstream, hence block the event-loop.

You could also play with emitOn, but fundamentally you will always have a need for worker threads.

[franck102]

I kind of understand what you are saying... but I still have a hard time wrapping my head around that programming model. I probably need to write a simple chain and debug through it to understand what is going on.

I have reverted to a good old blocking implementation for now, thanks for the help!

[jponge]

No worries, feel free to ping if you have further questions.

I'm sure you already have these references, but just in case:

• Mutiny has lots of Quarkus-agnostic guides: https://smallrye.io/smallrye-mutiny/2.3.1/
• Quarkus has reactive guides as well:
  • https://quarkus.io/guides/quarkus-reactive-architecture
  • https://quarkus.io/guides/getting-started-reactive
  • https://quarkus.io/guides/resteasy-reactive-migration