From 9a1978ec8c74b7e058cd8785d536a3184401d1d1 Mon Sep 17 00:00:00 2001 From: Saranya Somepalli Date: Fri, 13 Nov 2026 14:58:50 -0800 Subject: [PATCH] Split response interceptor stages into afterTransmission and modifyHttpResponse --- .../handler/BaseAsyncClientHandler.java | 15 +- .../internal/http/AmazonSyncHttpClient.java | 2 + ...sionInterceptorCallingResponseHandler.java | 40 +---- ...onseInterceptorCallingResponseHandler.java | 84 +++++++++ ...ransmissionExecutionInterceptorsStage.java | 29 +--- ...ttpResponseExecutionInterceptorsStage.java | 67 ++++++++ ...ntHandlerResponseInterceptorChainTest.java | 156 +++++++++++++++++ ...SyncHttpClientResponseInterceptorTest.java | 162 ++++++++++++++++++ 8 files changed, 499 insertions(+), 56 deletions(-) create mode 100644 core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/async/AsyncModifyHttpResponseInterceptorCallingResponseHandler.java create mode 100644 core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/ModifyHttpResponseExecutionInterceptorsStage.java create mode 100644 core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/handler/BaseAsyncClientHandlerResponseInterceptorChainTest.java create mode 100644 core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClientResponseInterceptorTest.java diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/handler/BaseAsyncClientHandler.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/handler/BaseAsyncClientHandler.java index 4e27f2400c18..782398be5421 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/handler/BaseAsyncClientHandler.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/handler/BaseAsyncClientHandler.java @@ -42,6 +42,7 @@ import software.amazon.awssdk.core.internal.http.IdempotentAsyncResponseHandler; import software.amazon.awssdk.core.internal.http.TransformingAsyncResponseHandler; import software.amazon.awssdk.core.internal.http.async.AsyncAfterTransmissionInterceptorCallingResponseHandler; +import software.amazon.awssdk.core.internal.http.async.AsyncModifyHttpResponseInterceptorCallingResponseHandler; import software.amazon.awssdk.core.internal.http.async.AsyncResponseHandler; import software.amazon.awssdk.core.internal.http.async.AsyncStreamingResponseHandler; import software.amazon.awssdk.core.internal.http.async.CombinedResponseAsyncHttpResponseHandler; @@ -231,8 +232,7 @@ private Comple finalizeSdkHttpRequestContext.asyncRequestBody().orElse(null), inputT, executionContext, - new AsyncAfterTransmissionInterceptorCallingResponseHandler<>(asyncResponseHandler, - executionContext)); + interceptorCallingResponseHandler(asyncResponseHandler, executionContext)); // Captured because the requestBody branch above may reassign 'marshalled', leaving it not effectively final. SdkHttpFullRequest requestForMetrics = marshalled; @@ -259,6 +259,17 @@ public void close() { client.close(); } + /** + * Wraps the response handler with the response interceptor steps, in order: read-only {@code afterTransmission} on + * the raw response, then the interceptors that can modify it. + */ + static TransformingAsyncResponseHandler interceptorCallingResponseHandler( + TransformingAsyncResponseHandler responseHandler, ExecutionContext executionContext) { + return new AsyncAfterTransmissionInterceptorCallingResponseHandler<>( + new AsyncModifyHttpResponseInterceptorCallingResponseHandler<>(responseHandler, executionContext), + executionContext); + } + /** * Error responses are never streaming so we always use {@link AsyncResponseHandler}. * diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClient.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClient.java index 53fc3c2e48a3..7ff37a78c59b 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClient.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClient.java @@ -46,6 +46,7 @@ import software.amazon.awssdk.core.internal.http.pipeline.stages.MakeRequestMutableStage; import software.amazon.awssdk.core.internal.http.pipeline.stages.MergeCustomHeadersStage; import software.amazon.awssdk.core.internal.http.pipeline.stages.MergeCustomQueryParamsStage; +import software.amazon.awssdk.core.internal.http.pipeline.stages.ModifyHttpResponseExecutionInterceptorsStage; import software.amazon.awssdk.core.internal.http.pipeline.stages.QueryParametersToBodyStage; import software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage; import software.amazon.awssdk.core.internal.http.pipeline.stages.SigningStage; @@ -197,6 +198,7 @@ public OutputT execute(HttpResponseHandler> response .then(BeforeTransmissionExecutionInterceptorsStage::new) .then(MakeHttpRequestStage::new) .then(AfterTransmissionExecutionInterceptorsStage::new) + .then(ModifyHttpResponseExecutionInterceptorsStage::new) .then(BeforeUnmarshallingExecutionInterceptorsStage::new) .then(() -> new HandleResponseStage<>(responseHandler)) .wrappedWith(ApiCallAttemptTimeoutTrackingStage::new) diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/async/AsyncAfterTransmissionInterceptorCallingResponseHandler.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/async/AsyncAfterTransmissionInterceptorCallingResponseHandler.java index 37f78e860479..e7cf9be77a21 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/async/AsyncAfterTransmissionInterceptorCallingResponseHandler.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/async/AsyncAfterTransmissionInterceptorCallingResponseHandler.java @@ -16,18 +16,17 @@ package software.amazon.awssdk.core.internal.http.async; import java.nio.ByteBuffer; -import java.util.Optional; import java.util.concurrent.CompletableFuture; import org.reactivestreams.Publisher; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.core.http.ExecutionContext; import software.amazon.awssdk.core.interceptor.InterceptorContext; import software.amazon.awssdk.core.internal.http.TransformingAsyncResponseHandler; -import software.amazon.awssdk.http.SdkHttpFullResponse; import software.amazon.awssdk.http.SdkHttpResponse; /** - * Async response handler decorator to run interceptors after response is received. + * Async response handler decorator that calls the read-only {@code afterTransmission} interceptors with the response + * exactly as received. * * @param the type of the result */ @@ -42,29 +41,18 @@ public AsyncAfterTransmissionInterceptorCallingResponseHandler(TransformingAsync this.context = context; } - private SdkHttpResponse beforeUnmarshalling(SdkHttpFullResponse response, ExecutionContext context) { + @Override + public void onHeaders(SdkHttpResponse response) { // Update interceptor context to include response - InterceptorContext interceptorContext = - context.interceptorContext().copy(b -> b.httpResponse(response)); + InterceptorContext interceptorContext = context.interceptorContext().copy(b -> b.httpResponse(response)); // interceptors.afterTransmission context.interceptorChain().afterTransmission(interceptorContext, context.executionAttributes()); - // interceptors.modifyHttpResponse - interceptorContext = context.interceptorChain().modifyHttpResponse(interceptorContext, context.executionAttributes()); - - // interceptors.beforeUnmarshalling - context.interceptorChain().beforeUnmarshalling(interceptorContext, context.executionAttributes()); - // Store updated context context.interceptorContext(interceptorContext); - return interceptorContext.httpResponse(); - } - - @Override - public void onHeaders(SdkHttpResponse response) { - delegate.onHeaders(beforeUnmarshalling((SdkHttpFullResponse) response, context)); // TODO: Ew + delegate.onHeaders(response); } @Override @@ -74,23 +62,11 @@ public void onError(Throwable error) { @Override public void onStream(Publisher publisher) { - Optional> newPublisher = context.interceptorChain() - .modifyAsyncHttpResponse(context.interceptorContext() - .toBuilder() - .responsePublisher(publisher) - .build(), - context.executionAttributes()) - .responsePublisher(); - - if (newPublisher.isPresent()) { - delegate.onStream(newPublisher.get()); - } else { - delegate.onStream(publisher); - } + delegate.onStream(publisher); } @Override public CompletableFuture prepare() { return delegate.prepare(); } -} \ No newline at end of file +} diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/async/AsyncModifyHttpResponseInterceptorCallingResponseHandler.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/async/AsyncModifyHttpResponseInterceptorCallingResponseHandler.java new file mode 100644 index 000000000000..11787579ada8 --- /dev/null +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/async/AsyncModifyHttpResponseInterceptorCallingResponseHandler.java @@ -0,0 +1,84 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.core.internal.http.async; + +import java.nio.ByteBuffer; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import org.reactivestreams.Publisher; +import software.amazon.awssdk.annotations.SdkInternalApi; +import software.amazon.awssdk.core.http.ExecutionContext; +import software.amazon.awssdk.core.interceptor.InterceptorContext; +import software.amazon.awssdk.core.internal.http.TransformingAsyncResponseHandler; +import software.amazon.awssdk.http.SdkHttpResponse; + +/** + * Async response handler decorator that calls the {@code modifyHttpResponse} and {@code beforeUnmarshalling} interceptors + * when headers arrive, and the {@code modifyAsyncHttpResponse} interceptors when the body arrives. + * + * @param the type of the result + */ +@SdkInternalApi +public final class AsyncModifyHttpResponseInterceptorCallingResponseHandler implements TransformingAsyncResponseHandler { + private final TransformingAsyncResponseHandler delegate; + private final ExecutionContext context; + + public AsyncModifyHttpResponseInterceptorCallingResponseHandler(TransformingAsyncResponseHandler delegate, + ExecutionContext context) { + this.delegate = delegate; + this.context = context; + } + + @Override + public void onHeaders(SdkHttpResponse response) { + // Update interceptor context to include response + InterceptorContext interceptorContext = context.interceptorContext().copy(b -> b.httpResponse(response)); + + // interceptors.modifyHttpResponse + interceptorContext = context.interceptorChain().modifyHttpResponse(interceptorContext, context.executionAttributes()); + + // interceptors.beforeUnmarshalling + context.interceptorChain().beforeUnmarshalling(interceptorContext, context.executionAttributes()); + + // Store updated context + context.interceptorContext(interceptorContext); + + delegate.onHeaders(interceptorContext.httpResponse()); + } + + @Override + public void onError(Throwable error) { + delegate.onError(error); + } + + @Override + public void onStream(Publisher publisher) { + Optional> newPublisher = context.interceptorChain() + .modifyAsyncHttpResponse(context.interceptorContext() + .toBuilder() + .responsePublisher(publisher) + .build(), + context.executionAttributes()) + .responsePublisher(); + + delegate.onStream(newPublisher.orElse(publisher)); + } + + @Override + public CompletableFuture prepare() { + return delegate.prepare(); + } +} diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java index 849f20a13689..6da273038d8d 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java @@ -21,11 +21,13 @@ import software.amazon.awssdk.core.internal.http.InterruptMonitor; import software.amazon.awssdk.core.internal.http.RequestExecutionContext; import software.amazon.awssdk.core.internal.http.pipeline.RequestPipeline; -import software.amazon.awssdk.http.AbortableInputStream; import software.amazon.awssdk.http.SdkHttpFullRequest; import software.amazon.awssdk.http.SdkHttpFullResponse; import software.amazon.awssdk.utils.Pair; +/** + * Calls the read-only {@code afterTransmission} interceptors with the response exactly as received. + */ @SdkInternalApi public class AfterTransmissionExecutionInterceptorsStage implements RequestPipeline, Pair> { @@ -33,36 +35,19 @@ public class AfterTransmissionExecutionInterceptorsStage public Pair execute(Pair input, RequestExecutionContext context) throws Exception { InterruptMonitor.checkInterrupted(input.right()); - InputStream originalResponseBody = input.right().content().orElse(null); + InputStream responseBody = input.right().content().orElse(null); // Update interceptor context InterceptorContext interceptorContext = context.executionContext().interceptorContext().copy(b -> b.httpResponse(input.right()) - .responseBody(originalResponseBody)); + .responseBody(responseBody)); // interceptors.afterTransmission context.interceptorChain().afterTransmission(interceptorContext, context.executionAttributes()); - // interceptors.modifyHttpResponse - interceptorContext = context.interceptorChain().modifyHttpResponse(interceptorContext, context.executionAttributes()); - - // When no replacement is returned, the response retains its original body. Keep the context aligned with the - // actual response so later interceptor failures can clean up that body. - if (!interceptorContext.responseBody().isPresent() && originalResponseBody != null) { - interceptorContext = interceptorContext.copy(b -> b.responseBody(originalResponseBody)); - } - - // Store updated context + // Store updated context, so later stages can clean up the body if they fail context.executionContext().interceptorContext(interceptorContext); - InterruptMonitor.checkInterrupted((SdkHttpFullResponse) interceptorContext.httpResponse()); - - SdkHttpFullResponse response = (SdkHttpFullResponse) interceptorContext.httpResponse(); - - if (interceptorContext.responseBody().isPresent()) { - response = response.toBuilder().content(AbortableInputStream.create(interceptorContext.responseBody().get())).build(); - } - - return Pair.of(input.left(), response); + return input; } } diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/ModifyHttpResponseExecutionInterceptorsStage.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/ModifyHttpResponseExecutionInterceptorsStage.java new file mode 100644 index 000000000000..3eaece763813 --- /dev/null +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/ModifyHttpResponseExecutionInterceptorsStage.java @@ -0,0 +1,67 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.core.internal.http.pipeline.stages; + +import java.io.InputStream; +import software.amazon.awssdk.annotations.SdkInternalApi; +import software.amazon.awssdk.core.interceptor.InterceptorContext; +import software.amazon.awssdk.core.internal.http.InterruptMonitor; +import software.amazon.awssdk.core.internal.http.RequestExecutionContext; +import software.amazon.awssdk.core.internal.http.pipeline.RequestPipeline; +import software.amazon.awssdk.http.AbortableInputStream; +import software.amazon.awssdk.http.SdkHttpFullRequest; +import software.amazon.awssdk.http.SdkHttpFullResponse; +import software.amazon.awssdk.utils.Pair; + +/** + * Calls the {@code modifyHttpResponse} interceptors and applies their changes to the response. + */ +@SdkInternalApi +public class ModifyHttpResponseExecutionInterceptorsStage + implements RequestPipeline, Pair> { + @Override + public Pair execute(Pair input, + RequestExecutionContext context) throws Exception { + InputStream responseBody = input.right().content().orElse(null); + + // Update interceptor context + InterceptorContext interceptorContext = + context.executionContext().interceptorContext().copy(b -> b.httpResponse(input.right()) + .responseBody(responseBody)); + + // interceptors.modifyHttpResponse + interceptorContext = context.interceptorChain().modifyHttpResponse(interceptorContext, context.executionAttributes()); + + // When no replacement is returned, the response retains its original body. Keep the context aligned with the + // actual response so later interceptor failures can clean up that body. + if (!interceptorContext.responseBody().isPresent() && responseBody != null) { + interceptorContext = interceptorContext.copy(b -> b.responseBody(responseBody)); + } + + // Store updated context + context.executionContext().interceptorContext(interceptorContext); + + InterruptMonitor.checkInterrupted((SdkHttpFullResponse) interceptorContext.httpResponse()); + + SdkHttpFullResponse response = (SdkHttpFullResponse) interceptorContext.httpResponse(); + + if (interceptorContext.responseBody().isPresent()) { + response = response.toBuilder().content(AbortableInputStream.create(interceptorContext.responseBody().get())).build(); + } + + return Pair.of(input.left(), response); + } +} diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/handler/BaseAsyncClientHandlerResponseInterceptorChainTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/handler/BaseAsyncClientHandlerResponseInterceptorChainTest.java new file mode 100644 index 000000000000..eb3367ed6372 --- /dev/null +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/handler/BaseAsyncClientHandlerResponseInterceptorChainTest.java @@ -0,0 +1,156 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.core.internal.handler; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.reactivex.Flowable; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Publisher; +import software.amazon.awssdk.core.http.ExecutionContext; +import software.amazon.awssdk.core.interceptor.Context; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptorChain; +import software.amazon.awssdk.core.interceptor.InterceptorContext; +import software.amazon.awssdk.core.internal.http.TransformingAsyncResponseHandler; +import software.amazon.awssdk.http.SdkHttpFullResponse; +import software.amazon.awssdk.http.SdkHttpResponse; +import utils.ValidSdkObjects; + +/** + * Verifies the order and effect of the async response interceptor chain built by {@link BaseAsyncClientHandler}. + */ +class BaseAsyncClientHandlerResponseInterceptorChainTest { + + private static final Publisher ORIGINAL_BODY = Flowable.just(utf8("original")); + private static final Publisher MODIFIED_BODY = Flowable.just(utf8("modified")); + + @Test + void responseChain_whenInterceptorModifiesResponse_afterTransmissionSeesOriginalAndDelegateSeesModified() { + RecordingInterceptor interceptor = new RecordingInterceptor(true); + CapturingHandler delegate = new CapturingHandler(); + + run(interceptor, delegate); + + assertThat(interceptor.calls) + .containsExactly("afterTransmission", "modifyHttpResponse", "beforeUnmarshalling", "modifyAsyncHttpResponse"); + assertThat(interceptor.afterTransmissionResponse.firstMatchingHeader("x-test")).hasValue("original"); + assertThat(interceptor.beforeUnmarshallingResponse.firstMatchingHeader("x-test")).hasValue("modified"); + assertThat(interceptor.modifyAsyncHttpResponsePublisher).isSameAs(ORIGINAL_BODY); + assertThat(delegate.headers.firstMatchingHeader("x-test")).hasValue("modified"); + assertThat(delegate.stream).isSameAs(MODIFIED_BODY); + } + + @Test + void responseChain_whenInterceptorDoesNotModifyResponse_delegateSeesOriginal() { + CapturingHandler delegate = new CapturingHandler(); + + run(new RecordingInterceptor(false), delegate); + + assertThat(delegate.headers.firstMatchingHeader("x-test")).hasValue("original"); + assertThat(delegate.stream).isSameAs(ORIGINAL_BODY); + } + + private static void run(ExecutionInterceptor interceptor, CapturingHandler delegate) { + ExecutionContext context = + ExecutionContext.builder() + .interceptorChain(new ExecutionInterceptorChain(Collections.singletonList(interceptor))) + .interceptorContext(InterceptorContext.builder().request(ValidSdkObjects.sdkRequest()).build()) + .executionAttributes(new ExecutionAttributes()) + .build(); + + TransformingAsyncResponseHandler chain = + BaseAsyncClientHandler.interceptorCallingResponseHandler(delegate, context); + chain.onHeaders(SdkHttpFullResponse.builder().statusCode(200).putHeader("x-test", "original").build()); + chain.onStream(ORIGINAL_BODY); + } + + private static ByteBuffer utf8(String s) { + return ByteBuffer.wrap(s.getBytes(StandardCharsets.UTF_8)); + } + + private static final class CapturingHandler implements TransformingAsyncResponseHandler { + private SdkHttpResponse headers; + private Publisher stream; + + @Override + public CompletableFuture prepare() { + return new CompletableFuture<>(); + } + + @Override + public void onHeaders(SdkHttpResponse headers) { + this.headers = headers; + } + + @Override + public void onStream(Publisher stream) { + this.stream = stream; + } + + @Override + public void onError(Throwable error) { + } + } + + private static final class RecordingInterceptor implements ExecutionInterceptor { + private final boolean modify; + private final List calls = new ArrayList<>(); + private SdkHttpResponse afterTransmissionResponse; + private SdkHttpResponse beforeUnmarshallingResponse; + private Publisher modifyAsyncHttpResponsePublisher; + + private RecordingInterceptor(boolean modify) { + this.modify = modify; + } + + @Override + public void afterTransmission(Context.AfterTransmission context, ExecutionAttributes executionAttributes) { + calls.add("afterTransmission"); + afterTransmissionResponse = context.httpResponse(); + } + + @Override + public SdkHttpResponse modifyHttpResponse(Context.ModifyHttpResponse context, + ExecutionAttributes executionAttributes) { + calls.add("modifyHttpResponse"); + return modify ? context.httpResponse().toBuilder().putHeader("x-test", "modified").build() + : context.httpResponse(); + } + + @Override + public void beforeUnmarshalling(Context.BeforeUnmarshalling context, ExecutionAttributes executionAttributes) { + calls.add("beforeUnmarshalling"); + beforeUnmarshallingResponse = context.httpResponse(); + } + + @Override + public Optional> modifyAsyncHttpResponseContent(Context.ModifyHttpResponse context, + ExecutionAttributes executionAttributes) { + calls.add("modifyAsyncHttpResponse"); + modifyAsyncHttpResponsePublisher = context.responsePublisher().orElse(null); + return modify ? Optional.of(MODIFIED_BODY) : context.responsePublisher(); + } + } +} diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClientResponseInterceptorTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClientResponseInterceptorTest.java new file mode 100644 index 000000000000..7652064ae6cb --- /dev/null +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClientResponseInterceptorTest.java @@ -0,0 +1,162 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.core.internal.http; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static software.amazon.awssdk.core.internal.util.ResponseHandlerTestUtils.combinedSyncResponseHandler; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.http.ExecutionContext; +import software.amazon.awssdk.core.http.NoopTestRequest; +import software.amazon.awssdk.core.interceptor.Context; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptorChain; +import software.amazon.awssdk.core.interceptor.InterceptorContext; +import software.amazon.awssdk.core.signer.NoOpSigner; +import software.amazon.awssdk.http.AbortableInputStream; +import software.amazon.awssdk.http.ExecutableHttpRequest; +import software.amazon.awssdk.http.HttpExecuteResponse; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.SdkHttpFullRequest; +import software.amazon.awssdk.http.SdkHttpResponse; +import software.amazon.awssdk.metrics.MetricCollector; +import software.amazon.awssdk.utils.IoUtils; +import software.amazon.awssdk.utils.Pair; +import utils.HttpTestUtils; +import utils.ValidSdkObjects; + +/** + * Runs a request through the full sync pipeline to verify the order and effect of the response interceptor hooks. + */ +class AmazonSyncHttpClientResponseInterceptorTest { + + private AmazonSyncHttpClient client; + + @BeforeEach + void setUp() throws Exception { + SdkHttpClient httpClient = mock(SdkHttpClient.class); + ExecutableHttpRequest call = mock(ExecutableHttpRequest.class); + when(httpClient.prepareRequest(any())).thenReturn(call); + when(httpClient.clientName()).thenReturn("UNKNOWN"); + SdkHttpResponse response = SdkHttpResponse.builder().statusCode(200).putHeader("x-test", "original").build(); + when(call.call()).thenReturn(HttpExecuteResponse.builder().response(response).responseBody(body("original")).build()); + client = HttpTestUtils.testClientBuilder().httpClient(httpClient).build(); + } + + @AfterEach + void tearDown() { + client.close(); + } + + @Test + void execute_whenInterceptorModifiesResponse_afterTransmissionSeesOriginalAndHandlerSeesModified() { + RecordingInterceptor interceptor = new RecordingInterceptor(true); + + Pair handled = execute(interceptor); + + assertThat(interceptor.calls).containsExactly("afterTransmission", "modifyHttpResponse", "beforeUnmarshalling"); + assertThat(interceptor.afterTransmissionResponse.firstMatchingHeader("x-test")).hasValue("original"); + assertThat(interceptor.beforeUnmarshallingResponse.firstMatchingHeader("x-test")).hasValue("modified"); + assertThat(handled.left().firstMatchingHeader("x-test")).hasValue("modified"); + assertThat(handled.right()).isEqualTo("modified"); + } + + @Test + void execute_whenInterceptorDoesNotModifyResponse_handlerSeesOriginal() { + Pair handled = execute(new RecordingInterceptor(false)); + + assertThat(handled.left().firstMatchingHeader("x-test")).hasValue("original"); + assertThat(handled.right()).isEqualTo("original"); + } + + private Pair execute(ExecutionInterceptor interceptor) { + SdkHttpFullRequest request = ValidSdkObjects.sdkHttpFullRequest().build(); + ExecutionContext executionContext = + ExecutionContext.builder() + .signer(new NoOpSigner()) + .interceptorChain(new ExecutionInterceptorChain(Collections.singletonList(interceptor))) + .executionAttributes(new ExecutionAttributes()) + .interceptorContext(InterceptorContext.builder() + .request(NoopTestRequest.builder().build()) + .httpRequest(request) + .build()) + .metricCollector(MetricCollector.create("ApiCall")) + .build(); + + return client.requestExecutionBuilder() + .request(request) + .originalRequest(NoopTestRequest.builder().build()) + .executionContext(executionContext) + .execute(combinedSyncResponseHandler( + (response, executionAttributes) -> + Pair.of((SdkHttpResponse) response, IoUtils.toUtf8String(response.content().get())), + null)); + } + + private static AbortableInputStream body(String content) { + return AbortableInputStream.create(new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8))); + } + + private static final class RecordingInterceptor implements ExecutionInterceptor { + private final boolean modify; + private final List calls = new ArrayList<>(); + private SdkHttpResponse afterTransmissionResponse; + private SdkHttpResponse beforeUnmarshallingResponse; + + private RecordingInterceptor(boolean modify) { + this.modify = modify; + } + + @Override + public void afterTransmission(Context.AfterTransmission context, ExecutionAttributes executionAttributes) { + calls.add("afterTransmission"); + afterTransmissionResponse = context.httpResponse(); + } + + @Override + public SdkHttpResponse modifyHttpResponse(Context.ModifyHttpResponse context, + ExecutionAttributes executionAttributes) { + calls.add("modifyHttpResponse"); + return modify ? context.httpResponse().toBuilder().putHeader("x-test", "modified").build() + : context.httpResponse(); + } + + @Override + public Optional modifyHttpResponseContent(Context.ModifyHttpResponse context, + ExecutionAttributes executionAttributes) { + return modify ? Optional.of(body("modified")) : context.responseBody(); + } + + @Override + public void beforeUnmarshalling(Context.BeforeUnmarshalling context, ExecutionAttributes executionAttributes) { + calls.add("beforeUnmarshalling"); + beforeUnmarshallingResponse = context.httpResponse(); + } + } +}