Return a Publisher<ByteBuffer> from streaming operations

[jhovell]

Expected Behavior

I would expect to not have to specify a Path as required in the method signature if I want to buffer contents to a stream

While this is fine:
CompletableFuture<GetObjectResponse> s3AsyncClient.getObject(GetObjectRequest, Path)

Proposed solution

I would also expect this:
CompletableFuture<GetObjectResponse> s3AsyncClient.getObject(GetObjectRequest)

I'm not sure where this fits in on a current roadmap if one exists, whether it was overlooked or is just not done yet. I would expect CompletableFuture to be a first class citizen, but maybe I have just misunderstood how it is meant to be used.

Your Environment

• AWS Java SDK version used: 2.0.0-preview-7
• JDK version used: Oracle Java(TM) SE Runtime Environment (build 9+181)
  Java HotSpot(TM) 64-Bit Server VM (build 9+181, mixed mode)
• Operating System and version: OSX

[millems]

InputStream is a little tricky with a non-blocking async client, because the operations on an InputStream are inherently blocking. Is your intent to handle the response in a blocking manner, and just use the async client for the request portion?

// Send the request asynchronously
s3AsyncClient.getObject(...).thenAccept(responseInputStream -> {
    // Handle the response in a synchronous/blocking manner
    IOUtils.copy(responseInputStream, System.out);
});

[erwan]

I have the same problem where I want the content for the s3 object, but I don't want to have to write it to a file.

If InputStream doesn't work, then maybe java.nio.Buffer would work better?

[hartmut-co-uk]

Hi @millems in my case I'd like to proxy the file via spring (reactor).
So I'd like to avoid 'buffering' to a file (FileAsyncResponseTransformer) or memory (ByteArrayAsyncResponseTransformer).

In my case I'd like to get the actual Publisher<ByteBuffer> - not an InputStream -> so I can could wrap it / use it as a Reactor Flux<ByteBuffer (or transformed Flux<DataBuffer>)

Example:

    fun download(request: ServerRequest): Mono<ServerResponse>: Mono<ServerResponse> {
        return Mono.fromFuture(
                client.getObject(
                        GetObjectRequest.builder()
                                .bucket(bucketName)
                                .key(key)
                                .build()
                )).flatMap {
                    ServerResponse.ok()
                            .contentType(MediaType.valueOf(mimeType))
                            .contentLength(contentLength)
                            .header("Content-Disposition", "attachment; filename=$filename")
                            .body(it.body)   // it := GetObjectResponse | GetObjectResponse.body := Publisher<ByteBuffer>
                }    
    }

Hope that makes sense...

Note:
I was trying to write my own AsyncResponseTransformer implementation to provide access to the publisher passed to AsyncResponseTransformer<ResponseT, ReturnT>.onStream(Publisher<ByteBuffer> publisher) - but what I ended up with was kind of a deadlock - in the way that the AsyncResponseTransformer.complete only is called once the 'publisher -> subscribe' has completed? - while the exposed publisher would only be subscribed to after the whole CompletableFuture<GetObjectResponse> will join - which again won't happen without the publisher subscribed to?

What would be the best approach to achieve what I'm trying to do here?

[millems]

@hartmut-co-uk It's a little tricky, since AsyncResponseTransformer.onStream can be called multiple times, in the event of a download failure followed by a retry. We only know if it won't be called again once complete is called, at which time you would have already needed to subscribe.

I suppose there could be a way to disable retries and expose the first publisher as soon as it is available. It's not as failure tolerate, but that's probably not the end of the world for many use-cases.

Would that fit your use-case?

[hartmut-co-uk]

Yes, I think so - many thanks for the immediate reply. 🙇

Would there be any alternative?

[millems]

Not sure. @shorea has thought more about the API than I have. Maybe he could weigh in?

[shorea]

@hartmut-co-uk can you explain your use case a bit more. You have access to the Publisher via the onStream method.

[millems]

@shorea The use-case as I understand it is to hand off access to the publisher:

public Publisher<?> getPublisher() {
    return s3.getObject(r -> r.bucket(...).key(...), ???);
}

Are you thinking it would just be done like this?

public Publisher<?> getPublisher() {
        CompletableFuture<Publisher<ByteBuffer>> publisherFuture = new CompletableFuture<>();

        s3.getObject(r -> r.bucket(...).key(...),
            new AsyncResponseTransformer<GetObjectResponse, Void>() {
                private boolean first = true;
                @Override
                public void responseReceived(StreamingOutputOperationResponse response) {

                }

                @Override
                public void onStream(SdkPublisher<ByteBuffer> publisher) {
                    if (!first) {
                        // Uh oh, another one! What do we do here?
                    }
                    publisherFuture.complete(publisher);
                    first = false;
                }

                @Override
                public void exceptionOccurred(Throwable throwable) {

                }

                @Override
                public Void complete() {
                    return null;
                }
            });
        return publisherFuture.join();
}

[shorea]

Why not just do what you need to do with the publisher in the onStream method?

[hartmut-co-uk]

@shorea as described above I'd like to proxy -> serve the object (response body).
(There's more things involved before that, authorization, db lookups, ...)

Anyway for me to proxy the content = response body - I need to use the publisher directly in my reactive chain -> to subscribe with my own app's ServerResponse. I don't think I'm able to do that from 'within' an AsyncResponseTransformer - as I need to wrap the CompletableFuture itself with a Mono.

I've had tried to expose the publisher with following implementation (in Kotlin...):

class FluxAsyncResponseTransformer : AsyncResponseTransformer<GetObjectResponse, ResponseByteBuffer<GetObjectResponse>> {

    private var response: GetObjectResponse? = null
    private var publisher: Publisher<ByteBuffer>? = null

    override fun responseReceived(response: GetObjectResponse) {
        this.response = response
    }

    override fun complete(): ResponseByteBuffer<GetObjectResponse> {
        return ResponseByteBuffer(response!!, publisher!!)
    }

    override fun onStream(publisher: Publisher<ByteBuffer>) {
        this.publisher = publisher
    }

    override fun exceptionOccurred(throwable: Throwable?) {
    }

}

But as far as I got with debugging - the complete method never is getting called - and my understanding was the client / getObject was waiting for the publisher to complete - before invoking the complete and completing the CompletableFuture of getObject.

So for me the whole reactive chain comes to a halt. Does that make sense?

[hartmut-co-uk]

Code reference:

aws-sdk-java-v2/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ResponseHandler.java

Line 255 in 34a6884

requestContext.handler().complete();

-> without a subscribe complete on the publisher - the CompletableFuture never completes..

So instead of trying to expose / get the publisher - I might need to pass/wrap the ServerResponse.ok().build(publisher) as a AsyncResponseTransformer like as a 'callback' maybe?

[creatorKoo]

I have same problem.
I can't not stream download in spring boot webflux.

Who is maintain this issue?