RestEasy reactive with javax.ws.rs.core.StreamingOutput

[swidersk...@gmail.com]

Hello,

While working on moving to rest easy reactive I ran into an issue with using javax.ws.rs.core.StreamingOutput to stream back a content of the response.

What is kind of surprising that the response code is 500 but there is nothing logged so no idea what might be missing. Or maybe this is not supported at all.

Here isa bit of code to illustrate how it is used

StreamingOutput entity = new ZipStreamingOutput(archived);

ResponseBuilder builder = Response.ok().entity(entity);

return builder.header("Content-Type", "application/zip").header("Content-Disposition",

        "attachment; filename=" + archived.getId() + ".zip").build();

And the ZipStreamOutput class looks like that

protected class ZipStreamingOutput implements StreamingOutput {

    private ArchivedItem archived;

    public ZipStreamingOutput(ArchivedItem archived) {

        this.archived = archived;

    }

    @Override

    public void write(OutputStream output) throws IOException, WebApplicationException {

        archived.writeAsZip(output);

    }

}

This works ok on rest easy classic and not sure I can do anything to workaround this, if you have any hints I’d appreciate.

Maciej

[Paul Carter-Brown]

Hi,

StreamingOutput is not supported currently on Resteasy Reactive. You can instead return a Multi ( I assume Multi<byte[]> should work ) and have ArchivedItem emit to the Multi as its writing as zip

--
You received this message because you are subscribed to the Google Groups "Quarkus Development mailing list" group.
To unsubscribe from this group and stop receiving emails from it, send an email to quarkus-dev...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/A1CC6569-BF84-4CA1-A35F-38ABDDDABBA9%40gmail.com.

[Stephane Epardaud]

What is this ArchivedItem type? This sounds like a blocking write in any case, so you'll need to do the writing on a worker thread with `@Blocking`.

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAO8YPTQkLq59%3Dk2BNutGJ7XzCZY%3DJJ2fqHXNASK1K7b-Sb7a0g%40mail.gmail.com.

--

Stéphane Épardaud

[Stephane Epardaud]

That's an interesting use-case. Given that you have all the bytes in memory, what's the rationale for not zipping them in memory too and returning a byte[] in the first place?

I'd assume that the zipped bytes would take less memory than the non-zipped ones, no?

In any case, I wonder if that end-point should be `@Blocking` or not. This is probably CPU-intensive and so might be better off on the worker thread, but I guess it depends if the CPU utilisation costs more than the offload to the worker pool. I guess you'd have to measure that, and also consider that doing compression on the IO thread might impact other endpoints that are not doing this compression (if you have more than one).

On Wed, 15 Sept 2021 at 10:32, Maciej Swiderski <swidersk...@gmail.com> wrote:

The ArchiveItem is a collection of file data already loaded (byte[]) so it is about to build a zip archive to deliver to the client. the writeToZip file creates zip entries and push them via ZipOutputStream.

Maciej

--

Stéphane Épardaud

[Paul Carter-Brown]

I'll do a quick POC of this and paste the code

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAKU9E9v7m48fFZZ%2B7%3DaCqjv%2B-8Y3LkFpK4UOwoZHzHAjnGkTgQ%40mail.gmail.com.

[Maciej Swiderski]

that's exactly what I rework it to. Generate the zip as byte[] and push it to the client. and yes, it is marked as blocking.

The use case is to deliver selected content as one zip archive to the client so zipping it was like most natural approach.

Maciej

[Maciej Swiderski]

thanks a lot Paul!

[Paul Carter-Brown]

An example of slowly replying with binary data in a stream using Resteasy reactive:

    @GET

    @Path("/multi-stream")
    @Produces(MediaType.APPLICATION_OCTET_STREAM)
    public Multi<byte[]> getSomeBinary(@QueryParam("approxBytes") int approxBytes) {
        log.error("In getMulti [{}]", approxBytes);
        Multi<byte[]> res = Multi.createFrom().emitter(emitter -> {
            Thread t = new Thread(() -> {
                log.error("Starting emitting data");
                int sent = 0;
                do {
                    byte[] data = getSomeSlowData();
                    sent += data.length;
                    emitter.emit(data);
                    log.error("Sent another [{}] bytes totalling [{}]", data.length, sent);
                } while (sent < approxBytes);
                emitter.complete();
                log.error("Finished emitting");
            });
            t.setName("data-emitter");
            t.start();
        });
        log.error("Out getMulti");
        return res;
    }

    private byte[] getSomeSlowData() {
        Random r = new Random();
        byte[] someData = new byte[10240];
        r.nextBytes(someData);
        try {
            Thread.sleep(1000);
        } catch (Exception e) {
            log.warn("Error:", e);
        }
        return someData;
    }

Your threading implementation in the emitter could use a pool or whatever makes sense. I'm just showing that the eventloop does the call but another thread that is free to block can be the one feeding the data back. Here are the logs showing what threads are involved:

2021-09-15 10:07:19,539 ERROR [gur.jin.ser.ser.res.ScenariosResource] (vert.x-eventloop-thread-19) In getMulti [100000]
2021-09-15 10:07:19,539 ERROR [gur.jin.ser.ser.res.ScenariosResource] (vert.x-eventloop-thread-19) Out getMulti
2021-09-15 10:07:19,540 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Starting emitting data
2021-09-15 10:07:20,541 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [10240]
2021-09-15 10:07:21,542 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [20480]
2021-09-15 10:07:22,543 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [30720]
2021-09-15 10:07:23,544 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [40960]
2021-09-15 10:07:24,546 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [51200]
2021-09-15 10:07:25,547 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [61440]
2021-09-15 10:07:26,548 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [71680]
2021-09-15 10:07:27,550 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [81920]
2021-09-15 10:07:28,551 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [92160]
2021-09-15 10:07:29,552 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Sent another [10240] bytes totalling [102400]
2021-09-15 10:07:29,553 ERROR [gur.jin.ser.ser.res.ScenariosResource] (data-emitter) Finished emitting

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAFA9VrGA36qcgOXaa1igaUGsPCQTK0oNP7opqtSD62%3DrirGB1w%40mail.gmail.com.

[Maciej Swiderski]

Thanks Paul,

it is bit different than what I need but certainly helps to better understand how to use Multi

Maciej

[Stephane Epardaud]

Thanks Paul. However I recommend using the Quarkus worker pool if you want to do that:

    @Inject
    ManagedExecutor workerExecutor;

    @GET
    @Path("/multi-stream")
    @Produces(MediaType.APPLICATION_OCTET_STREAM)

    public Multi<byte[]> getSomeBinary(@RestQuery int approxBytes) {

        log.error("In getMulti [{}]", approxBytes);

        Multi<byte[]> res = Multi.createFrom().<byte[]>emitter(emitter -> {

            log.error("Starting emitting data");
            int sent = 0;
            do {
                byte[] data = getSomeSlowData();
                sent += data.length;
                emitter.emit(data);
                log.error("Sent another [{}] bytes totalling [{}]", data.length, sent);
            } while (sent < approxBytes);
            emitter.complete();
            log.error("Finished emitting");

        }).runSubscriptionOn(workerExecutor);

        log.error("Out getMulti");
        return res;
    }

    private byte[] getSomeSlowData() {
        Random r = new Random();
        byte[] someData = new byte[10240];
        r.nextBytes(someData);
        try {
            Thread.sleep(1000);
        } catch (Exception e) {
            log.warn("Error:", e);
        }
        return someData;
    }

Alternately, you can also just mark the endpoint as blocking, and both the endpoint and the subscription will run on the worker thread pool:

    @Blocking

    @GET
    @Path("/multi-stream")
    @Produces(MediaType.APPLICATION_OCTET_STREAM)

    public Multi<byte[]> getSomeBinary(@RestQuery int approxBytes) {

        log.error("In getMulti [{}]", approxBytes);

        Multi<byte[]> res = Multi.createFrom().<byte[]>emitter(emitter -> {

            log.error("Starting emitting data");
            int sent = 0;
            do {
                byte[] data = getSomeSlowData();
                sent += data.length;
                emitter.emit(data);
                log.error("Sent another [{}] bytes totalling [{}]", data.length, sent);
            } while (sent < approxBytes);
            emitter.complete();
            log.error("Finished emitting");
        });

--

Stéphane Épardaud