Line-by-line publisher of S3 object content

[martin-tarjanyi]

Does the current version of the SDK support returning the content of an S3 object reactively in a Publisher line by line similar to for example Athena pagination?

[zoewangg]

You can use S3AsyncClient and provide your own AsyncResponseTransformer.

Example:

aws-sdk-java-v2/services/s3/src/it/java/software/amazon/awssdk/services/s3/GetObjectAsyncIntegrationTest.java

Line 104 in 31a1767

.getObject(getObjectRequest, new AsyncResponseTransformer<GetObjectResponse, String>() {

[martin-tarjanyi]

I see. And do you have plan to add an async response transfomer which exposes the ByteBuffer publisher?

I'm thinking about something like this:

public class ByteBufferSdkPublisherAsyncResponseTransformer<T extends SdkResponse>
        implements AsyncResponseTransformer<T, PublisherWrapper<T, ByteBuffer>>
{
    private volatile CompletableFuture<SdkPublisher<ByteBuffer>> cf;
    private volatile T response;

    @Override
    public CompletableFuture<PublisherWrapper<T, ByteBuffer>> prepare()
    {
        cf = new CompletableFuture<>();

        return cf.thenApply(publisher -> new PublisherWrapper<>(publisher, response));
    }

    @Override
    public void onResponse(T response)
    {
        this.response = response;
    }

    @Override
    public void onStream(SdkPublisher<ByteBuffer> publisher)
    {
        cf.complete(publisher);
    }

    @Override
    public void exceptionOccurred(Throwable error)
    {
        cf.completeExceptionally(error);
    }
}

Although, the ByteBuffer inside the CompletableFuture is not very nice to say the least. Instead I would imagine a method (getObjectByteBufferStream) in the S3AsyncClient class which could return SdkPublisher<ByteArray>.

What do you think?

[zoewangg]

Hmm, can you tell us a bit more of your use case? We might consider adding it if that's a common use case.

[martin-tarjanyi]

Sure.
I have a text file in s3 which contains separate/individual data in each line like this:

useful information 1
useful information 2
useful information 3
...

The file is quite big, so I wouldn't like to load the whole file into memory. Instead, I would like to process it reactively line by line with backpressure.

[iconara]

I'm struggling with this too, I have code that does some S3 operations that I want to be able to do asynchronously, but for some of them the result needs to be handed off to for example Commons CSV that expects a Reader. I have written an AsyncResponseTransformer implementation that uses a LinkedBlockingQueue internally to present a InputStream interface, but I'm struggling with how to use the API correctly.

Specifically, when subscribing and requesting more, what is the unit of the Subscription#request call? Is it bytes? Is it calls to #onNext? Is there any way to know or control the size of the chunks passed to #onNext? I haven't been able to figure this out.

For me, it would obviously be much easier if there was an SDK implementation that would give me an InputStream or Reader from an asynchronous call, or like @zoewangg is requesting, a Publisher that managed buffering and produced lines, or custom sized chunks.

[deepankerk]

Does the current version of the SDK support returning the content of an S3 object reactively in a Publisher line by line similar to for example Athena pagination?

I was able to achieve line by line streaming using the below code:

ResponseInputStream responseInputStream = client.getObject(GetObjectRequest.builder()
.bucket(s3Entity.getBucket().getName())
.key(s3Entity.getObject().getKey())
.build(), ResponseTransformer.toInputStream());

int linesRead = 0;
try (BufferedReader reader =
new BufferedReader(new InputStreamReader(responseInputStream, StandardCharsets.UTF_8))) {
while (reader.readLine() != null) {
linesRead++;
}
} catch (IOException e) {
logger.log("IO Exception: "+e);
}
logger.log("Lines read: "+linesRead);`

This is not async though. I stumbled onto this issue looking for a better way to parse S3 files. More details here: #1358

[JosefGold]

We have this use case as well, would be happy to have an implementation in the SDK

[koziolk]

@zoewangg Any update on when the feature described by @martin-tarjanyi will be added?

[millems]

Not at this time, but it's in our backlog. A PR is much easier to prioritize, if someone wanted to submit one for this feature.

[Bennett-Lynch]

Hi all,

We just released support for a new AsyncResponseTransformer: toPublisher(). This transformer makes it easier to directly consume a streaming-response payload (e.g., S3 GetObject) with async clients.

It returns a ResponsePublisher, which contains a reference to the SdkResponse returned by the service and a SdkPublisher<ByteBuffer>, which allows you to directly consume the streaming content.

This transformer is available starting in version 2.17.122. For more details and example usage, please see the Javadoc:

aws-sdk-java-v2/core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncResponseTransformer.java

Lines 147 to 176 in 0e00a99

	/**
	* Creates an {@link AsyncResponseTransformer} that publishes the response body content through a {@link ResponsePublisher},
	* which is an {@link SdkPublisher} that also contains a reference to the {@link SdkResponse} returned by the service.
	* <p>
	* When this transformer is used with an async client, the {@link CompletableFuture} that the client returns will be completed
	* once the {@link SdkResponse} is available and the response body <i>begins</i> streaming. This behavior differs from some
	* other transformers, like {@link #toFile(Path)} and {@link #toBytes()}, which only have their {@link CompletableFuture}
	* completed after the entire response body has finished streaming.
	* <p>
	* You are responsible for subscribing to this publisher and managing the associated back-pressure. Therefore, this
	* transformer is only recommended for advanced use cases.
	* <p>
	* Example usage:
	* <pre>
	* {@code
	* CompletableFuture<ResponsePublisher<GetObjectResponse>> responseFuture =
	* s3AsyncClient.getObject(getObjectRequest, AsyncResponseTransformer.toPublisher());
	* ResponsePublisher<GetObjectResponse> responsePublisher = responseFuture.join();
	* System.out.println(responsePublisher.response());
	* CompletableFuture<Void> drainPublisherFuture = responsePublisher.subscribe(System.out::println);
	* drainPublisherFuture.join();
	* }
	* </pre>
	*
	* @param <ResponseT> Pojo response type.
	* @return AsyncResponseTransformer instance.
	*/
	static <ResponseT extends SdkResponse> AsyncResponseTransformer<ResponseT, ResponsePublisher<ResponseT>> toPublisher() {
	return new PublisherAsyncResponseTransformer<>();
	}

The ByteBuffer-stream could then be transformed into strings as follows:

SdkPublisher<ByteBuffer> bytePublisher = ...
SdkPublisher<String> strPublisher = bytePublisher.map(buf -> StandardCharsets.UTF_8.decode(buf).toString());

However, this does not buffer and split into lines. You would need to implement that transformation on your own.

Please let us know if you find a SdkPublisher<String> sufficient for this feature request or if you would still like to see an included line-by-line publisher. I think this would be a reasonable feature to include, as it would allow reading an S3 object similar to how the JDK allows reading a local file with the Files#lines(Path) API.