Add new AsyncResponseTransformer: toPublisher()

[Bennett-Lynch]

Description

Add new AsyncResponseTransformer: toPublisher(). This transformer makes it more convenient for users to directly consume a streaming-response payload (i.e., S3 GetObject) with async clients. This also allows users of Reactor/RxJava to more easily consume a streaming response (e.g., via Flux#from(Publisher)).

Motivation and Context

AsyncResponseTransformer currently lacks a static factory method that would allow a user to stream the response body data. The only methods currently available are toFile() and toBytes(). This lacks parity when compared to AsyncRequestBody#fromPublisher(), which allows declaring a request body from an arbitrary Publisher<ByteBuffer>. It also lacks parity when compared to the sync client's response equivalent, ResponseTransformer#toInputStream(), which streams a response body to an InputStream.

Related Issues

1. Return a Publisher<ByteBuffer> from streaming operations #391
2. Line-by-line publisher of S3 object content #1253
3. Support S3TransferManager upload using Flux<ByteBuffer> #2714

Modifications

1. Create new AsyncResponseTransformer#toPublisher() static factory method that publishes the response body content through a ResponsePublisher, which is an SdkPublisher that also contains a reference to the SdkResponse returned by the service.
2. Update generated async clients' response-streaming operations to publish metrics explicitly at end-of-stream, rather than end-of-future. This is necessary because the provided transformer implementation determines when the associated future is completed. For to-file and to-bytes transformers, it is not completed until the end of the response body. But for the new to-publisher transformer, the future is completed when receiving the response header.
3. To facilitate the above requirement, move NotifyingAsyncResponseTransformer from being a TM-internal-API to a SDK-protected-API. Refactor NotifyingAsyncResponseTransformer into 3 separate listeners interfaces: AsyncResponseTransformerListener, PublisherListener, SubscriberListener. Each listener type allows wrapping the associated interface to listen in on events during the request/response lifecycle. Also convert NotifyingAsyncRequestBody to SDK-protected AsyncRequestBodyListener.
4. Update TransferManager's TransferProgressUpdater to use the new transformer-listener logic to notify of a completedTransfer only after stream completion. Update TransferManager's TransferListenerInvoker to guarantee one-time callback execution, now that there are multiple possible origins of success and failure.
5. Add integration tests for TransferManager and S3AsyncClient using the new toPublisher feature.

License

• I confirm that this pull request can be released under the Apache 2 license

[sonarqubecloud]

SonarCloud Quality Gate failed.   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
3 Code Smells

37.8% Coverage
0.0% Duplication

[Bennett-Lynch]

@pkgonan Let me know if you think this would be useful. It's related to #2714, but for downloads rather than uploads. It's possible to implement your own AsyncResponseTransformer to do this today, but offering it as part of the SDK should help simplify that use case.

[pkgonan]

@Bennett-Lynch
If there is an upload, there must also be a download. Thanks a lot. It looks very useful.

[pkgonan]

@Bennett-Lynch
Hi. When do you expect this feature done?

[Bennett-Lynch]

Hi @pkgonan. I don't have an estimate to share at this time, but if this is a feature you could benefit from, please let me know and I will try to find cycles to prioritize it.

In the meantime, if you need the functionality immediately, you should be able to copy the classes that were introduced in this PR. These changes don't require any substantial changes to the inner workings of the SDK and instead are a mostly standalone and pluggable implementation of the AsyncResponseTransformer interface. I understand copying classes is not an ideal solution, but I just wanted to mention it as a temporary solution.

[pkgonan]

@Bennett-Lynch Hi. Yes I need this feature. We are considering this feature in new projects. Therefore, I was constantly checking the progress of this PR.

[zoewangg]

Can we make sure integ tests pass before merging?

[sonarqubecloud]

SonarCloud Quality Gate failed.   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
10 Code Smells

87.1% Coverage
3.7% Duplication

[Bennett-Lynch]

This change has been released as part of version 2.17.122.

[pkgonan]

@Bennett-Lynch
Hi.

When using this feature, the LeasedConcurrency metric will continue to rise.

Perhaps there is a leak in your connection.