|
34 | 34 | import software.amazon.awssdk.core.http.Crc32Validation; |
35 | 35 | import software.amazon.awssdk.core.http.ExecutionContext; |
36 | 36 | import software.amazon.awssdk.core.http.HttpResponseHandler; |
| 37 | +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; |
37 | 38 | import software.amazon.awssdk.core.interceptor.InterceptorContext; |
38 | 39 | import software.amazon.awssdk.core.internal.InternalCoreExecutionAttribute; |
39 | 40 | import software.amazon.awssdk.core.internal.http.AmazonAsyncHttpClient; |
@@ -88,51 +89,47 @@ public <InputT extends SdkRequest, OutputT extends SdkResponse, ReturnT> Complet |
88 | 89 | AsyncResponseTransformer<OutputT, ReturnT> asyncResponseTransformer) { |
89 | 90 |
|
90 | 91 | return measureApiCallSuccess(executionParams, () -> { |
91 | | - // Running beforeExecution interceptors and modifyRequest interceptors. |
92 | | - ExecutionContext context = invokeInterceptorsAndCreateExecutionContext(executionParams); |
| 92 | + if (executionParams.getCombinedResponseHandler() != null) { |
| 93 | + // There is no support for catching errors in a body for streaming responses. Our codegen must never |
| 94 | + // attempt to do this. |
| 95 | + throw new IllegalArgumentException("A streaming 'asyncResponseTransformer' may not be used when a " |
| 96 | + + "'combinedResponseHandler' has been specified in a " |
| 97 | + + "ClientExecutionParams object."); |
| 98 | + } |
93 | 99 |
|
94 | | - TransformingAsyncResponseHandler<Response<ReturnT>> combinedResponseHandler = |
95 | | - createStreamingCombinedResponseHandler(executionParams, asyncResponseTransformer, context); |
| 100 | + ExecutionAttributes executionAttributes = executionParams.executionAttributes(); |
| 101 | + executionAttributes.putAttribute(InternalCoreExecutionAttribute.EXECUTION_ATTEMPT, 1); |
96 | 102 |
|
97 | | - return doExecute(executionParams, context, combinedResponseHandler); |
98 | | - }); |
99 | | - } |
| 103 | + AsyncStreamingResponseHandler<OutputT, ReturnT> asyncStreamingResponseHandler = |
| 104 | + new AsyncStreamingResponseHandler<>(asyncResponseTransformer); |
100 | 105 |
|
101 | | - private <InputT extends SdkRequest, OutputT extends SdkResponse, ReturnT> TransformingAsyncResponseHandler<Response<ReturnT>> |
102 | | - createStreamingCombinedResponseHandler(ClientExecutionParams<InputT, OutputT> executionParams, |
103 | | - AsyncResponseTransformer<OutputT, ReturnT> asyncResponseTransformer, |
104 | | - ExecutionContext context) { |
105 | | - if (executionParams.getCombinedResponseHandler() != null) { |
106 | | - // There is no support for catching errors in a body for streaming responses. Our codegen must never |
107 | | - // attempt to do this. |
108 | | - throw new IllegalArgumentException("A streaming 'asyncResponseTransformer' may not be used when a " |
109 | | - + "'combinedResponseHandler' has been specified in a " |
110 | | - + "ClientExecutionParams object."); |
111 | | - } |
| 106 | + // For streaming requests, prepare() should be called as early as possible to avoid NPE in client |
| 107 | + // See https://github.com/aws/aws-sdk-java-v2/issues/1268. We do this with a wrapper that caches the prepare |
| 108 | + // result until the execution attempt number changes. This guarantees that prepare is only called once per |
| 109 | + // execution. |
| 110 | + TransformingAsyncResponseHandler<ReturnT> wrappedAsyncStreamingResponseHandler = |
| 111 | + IdempotentAsyncResponseHandler.create( |
| 112 | + asyncStreamingResponseHandler, |
| 113 | + () -> executionAttributes.getAttribute(InternalCoreExecutionAttribute.EXECUTION_ATTEMPT), |
| 114 | + Integer::equals); |
| 115 | + wrappedAsyncStreamingResponseHandler.prepare(); |
112 | 116 |
|
113 | | - AsyncStreamingResponseHandler<OutputT, ReturnT> asyncStreamingResponseHandler = |
114 | | - new AsyncStreamingResponseHandler<>(asyncResponseTransformer); |
| 117 | + // Running beforeExecution interceptors and modifyRequest interceptors. |
| 118 | + ExecutionContext context = invokeInterceptorsAndCreateExecutionContext(executionParams); |
115 | 119 |
|
116 | | - // For streaming requests, prepare() should be called as early as possible to avoid NPE in client |
117 | | - // See https://github.com/aws/aws-sdk-java-v2/issues/1268. We do this with a wrapper that caches the prepare |
118 | | - // result until the execution attempt number changes. This guarantees that prepare is only called once per |
119 | | - // execution. |
120 | | - TransformingAsyncResponseHandler<ReturnT> wrappedAsyncStreamingResponseHandler = |
121 | | - IdempotentAsyncResponseHandler.create( |
122 | | - asyncStreamingResponseHandler, |
123 | | - () -> context.executionAttributes().getAttribute(InternalCoreExecutionAttribute.EXECUTION_ATTEMPT), |
124 | | - Integer::equals); |
125 | | - wrappedAsyncStreamingResponseHandler.prepare(); |
| 120 | + HttpResponseHandler<OutputT> decoratedResponseHandlers = |
| 121 | + decorateResponseHandlers(executionParams.getResponseHandler(), context); |
126 | 122 |
|
127 | | - HttpResponseHandler<OutputT> decoratedResponseHandlers = |
128 | | - decorateResponseHandlers(executionParams.getResponseHandler(), context); |
| 123 | + asyncStreamingResponseHandler.responseHandler(decoratedResponseHandlers); |
129 | 124 |
|
130 | | - asyncStreamingResponseHandler.responseHandler(decoratedResponseHandlers); |
| 125 | + TransformingAsyncResponseHandler<? extends SdkException> errorHandler = |
| 126 | + resolveErrorResponseHandler(executionParams.getErrorResponseHandler(), context, crc32Validator); |
131 | 127 |
|
132 | | - TransformingAsyncResponseHandler<? extends SdkException> errorHandler = |
133 | | - resolveErrorResponseHandler(executionParams.getErrorResponseHandler(), context, crc32Validator); |
| 128 | + TransformingAsyncResponseHandler<Response<ReturnT>> combinedResponseHandler = |
| 129 | + new CombinedResponseAsyncHttpResponseHandler<>(wrappedAsyncStreamingResponseHandler, errorHandler); |
134 | 130 |
|
135 | | - return new CombinedResponseAsyncHttpResponseHandler<>(wrappedAsyncStreamingResponseHandler, errorHandler); |
| 131 | + return doExecute(executionParams, context, combinedResponseHandler); |
| 132 | + }); |
136 | 133 | } |
137 | 134 |
|
138 | 135 | private <InputT extends SdkRequest, OutputT extends SdkResponse> TransformingAsyncResponseHandler<Response<OutputT>> |
|
0 commit comments