|
| 1 | +/* |
| 2 | + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. |
| 3 | + * |
| 4 | + * Licensed under the Apache License, Version 2.0 (the "License"). |
| 5 | + * You may not use this file except in compliance with the License. |
| 6 | + * A copy of the License is located at |
| 7 | + * |
| 8 | + * http://aws.amazon.com/apache2.0 |
| 9 | + * |
| 10 | + * or in the "license" file accompanying this file. This file is distributed |
| 11 | + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either |
| 12 | + * express or implied. See the License for the specific language governing |
| 13 | + * permissions and limitations under the License. |
| 14 | + */ |
| 15 | + |
| 16 | +package software.amazon.awssdk.services.mediastoredata; |
| 17 | + |
| 18 | +import static org.assertj.core.api.AssertionsForClassTypes.assertThat; |
| 19 | +import static software.amazon.awssdk.utils.FunctionalUtils.invokeSafely; |
| 20 | + |
| 21 | +import io.reactivex.Flowable; |
| 22 | +import java.io.ByteArrayInputStream; |
| 23 | +import java.io.FilterInputStream; |
| 24 | +import java.io.IOException; |
| 25 | +import java.io.InputStream; |
| 26 | +import java.net.URI; |
| 27 | +import java.nio.ByteBuffer; |
| 28 | +import java.nio.charset.StandardCharsets; |
| 29 | +import java.time.Duration; |
| 30 | +import java.time.Instant; |
| 31 | +import java.util.ArrayList; |
| 32 | +import java.util.List; |
| 33 | +import java.util.Optional; |
| 34 | +import org.apache.commons.lang3.RandomStringUtils; |
| 35 | +import org.junit.jupiter.api.AfterAll; |
| 36 | +import org.junit.jupiter.api.BeforeAll; |
| 37 | +import org.junit.jupiter.api.Test; |
| 38 | +import org.reactivestreams.Subscriber; |
| 39 | +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; |
| 40 | +import software.amazon.awssdk.core.async.AsyncRequestBody; |
| 41 | +import software.amazon.awssdk.core.interceptor.Context; |
| 42 | +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; |
| 43 | +import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; |
| 44 | +import software.amazon.awssdk.core.sync.RequestBody; |
| 45 | +import software.amazon.awssdk.http.ContentStreamProvider; |
| 46 | +import software.amazon.awssdk.http.apache.ApacheHttpClient; |
| 47 | +import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient; |
| 48 | +import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient; |
| 49 | +import software.amazon.awssdk.services.mediastore.MediaStoreClient; |
| 50 | +import software.amazon.awssdk.services.mediastore.model.Container; |
| 51 | +import software.amazon.awssdk.services.mediastore.model.ContainerStatus; |
| 52 | +import software.amazon.awssdk.services.mediastore.model.DescribeContainerResponse; |
| 53 | +import software.amazon.awssdk.services.mediastoredata.model.DeleteObjectRequest; |
| 54 | +import software.amazon.awssdk.services.mediastoredata.model.ObjectNotFoundException; |
| 55 | +import software.amazon.awssdk.services.mediastoredata.model.PutObjectRequest; |
| 56 | +import software.amazon.awssdk.testutils.Waiter; |
| 57 | +import software.amazon.awssdk.testutils.service.AwsIntegrationTestBase; |
| 58 | + |
| 59 | +/** |
| 60 | + * Integration test to verify Transfer-Encoding:chunked functionalities for all supported HTTP clients. Do not delete. |
| 61 | + */ |
| 62 | +public class TransferEncodingChunkedIntegrationTest extends AwsIntegrationTestBase { |
| 63 | + private static final String CONTAINER_NAME = "java-sdk-test-" + Instant.now().toEpochMilli(); |
| 64 | + private static MediaStoreClient mediaStoreClient; |
| 65 | + private static MediaStoreDataClient syncClientWithApache; |
| 66 | + private static MediaStoreDataClient syncClientWithUrlConnection; |
| 67 | + private static MediaStoreDataAsyncClient asyncClientWithNetty; |
| 68 | + private static AwsCredentialsProvider credentialsProvider; |
| 69 | + private static Container container; |
| 70 | + private static PutObjectRequest putObjectRequest; |
| 71 | + private static DeleteObjectRequest deleteObjectRequest; |
| 72 | + |
| 73 | + @BeforeAll |
| 74 | + public static void setup() { |
| 75 | + credentialsProvider = getCredentialsProvider(); |
| 76 | + mediaStoreClient = MediaStoreClient.builder() |
| 77 | + .credentialsProvider(credentialsProvider) |
| 78 | + .httpClient(ApacheHttpClient.builder().build()) |
| 79 | + .build(); |
| 80 | + container = createContainer(); |
| 81 | + URI uri = URI.create(container.endpoint()); |
| 82 | + |
| 83 | + syncClientWithApache = MediaStoreDataClient.builder() |
| 84 | + .endpointOverride(uri) |
| 85 | + .credentialsProvider(credentialsProvider) |
| 86 | + .httpClient(ApacheHttpClient.builder().build()) |
| 87 | + .overrideConfiguration(o -> o.addExecutionInterceptor(new CaptureTransferEncodingHeaderInterceptor())) |
| 88 | + .build(); |
| 89 | + |
| 90 | + syncClientWithUrlConnection= MediaStoreDataClient.builder() |
| 91 | + .endpointOverride(uri) |
| 92 | + .credentialsProvider(credentialsProvider) |
| 93 | + .httpClient(UrlConnectionHttpClient.create()) |
| 94 | + .overrideConfiguration(o -> o.addExecutionInterceptor(new CaptureTransferEncodingHeaderInterceptor())) |
| 95 | + .build(); |
| 96 | + |
| 97 | + asyncClientWithNetty = MediaStoreDataAsyncClient.builder() |
| 98 | + .endpointOverride(uri) |
| 99 | + .credentialsProvider(getCredentialsProvider()) |
| 100 | + .httpClient(NettyNioAsyncHttpClient.create()) |
| 101 | + .overrideConfiguration(o -> o.addExecutionInterceptor(new CaptureTransferEncodingHeaderInterceptor())) |
| 102 | + .build(); |
| 103 | + |
| 104 | + putObjectRequest = PutObjectRequest.builder() |
| 105 | + .contentType("application/octet-stream") |
| 106 | + .path("/foo") |
| 107 | + .build(); |
| 108 | + |
| 109 | + deleteObjectRequest = DeleteObjectRequest.builder() |
| 110 | + .path("/foo") |
| 111 | + .build(); |
| 112 | + } |
| 113 | + |
| 114 | + @AfterAll |
| 115 | + public static void tearDown() { |
| 116 | + syncClientWithApache.deleteObject(deleteObjectRequest); |
| 117 | + Waiter.run(() -> syncClientWithApache.describeObject(r -> r.path("/foo"))) |
| 118 | + .untilException(ObjectNotFoundException.class) |
| 119 | + .orFailAfter(Duration.ofMinutes(1)); |
| 120 | + CaptureTransferEncodingHeaderInterceptor.reset(); |
| 121 | + } |
| 122 | + |
| 123 | + @Test |
| 124 | + public void apacheClientPutObject_withoutContentLength_sendsSuccessfully() { |
| 125 | + TestContentProvider provider = new TestContentProvider(RandomStringUtils.random(1000).getBytes(StandardCharsets.UTF_8)); |
| 126 | + syncClientWithApache.putObject(putObjectRequest, RequestBody.fromContentProvider(provider, "binary/octet-stream")); |
| 127 | + assertThat(CaptureTransferEncodingHeaderInterceptor.isChunked).isTrue(); |
| 128 | + } |
| 129 | + |
| 130 | + @Test |
| 131 | + public void urlConnectionClientPutObject_withoutContentLength_sendsSuccessfully() { |
| 132 | + TestContentProvider provider = new TestContentProvider(RandomStringUtils.random(1000).getBytes(StandardCharsets.UTF_8)); |
| 133 | + syncClientWithUrlConnection.putObject(putObjectRequest, RequestBody.fromContentProvider(provider, "binary/octet-stream")); |
| 134 | + assertThat(CaptureTransferEncodingHeaderInterceptor.isChunked).isTrue(); |
| 135 | + } |
| 136 | + |
| 137 | + @Test |
| 138 | + public void nettyClientPutObject_withoutContentLength_sendsSuccessfully() { |
| 139 | + asyncClientWithNetty.putObject(putObjectRequest, customAsyncRequestBodyWithoutContentLength()).join(); |
| 140 | + assertThat(CaptureTransferEncodingHeaderInterceptor.isChunked).isTrue(); |
| 141 | + } |
| 142 | + |
| 143 | + private static Container createContainer() { |
| 144 | + mediaStoreClient.createContainer(r -> r.containerName(CONTAINER_NAME)); |
| 145 | + DescribeContainerResponse response = waitContainerToBeActive(); |
| 146 | + return response.container(); |
| 147 | + } |
| 148 | + |
| 149 | + private static DescribeContainerResponse waitContainerToBeActive() { |
| 150 | + return Waiter.run(() -> mediaStoreClient.describeContainer(r -> r.containerName(CONTAINER_NAME))) |
| 151 | + .until(r -> ContainerStatus.ACTIVE.equals(r.container().status())) |
| 152 | + .orFailAfter(Duration.ofMinutes(3)); |
| 153 | + } |
| 154 | + |
| 155 | + private static class CaptureTransferEncodingHeaderInterceptor implements ExecutionInterceptor { |
| 156 | + private static boolean isChunked; |
| 157 | + |
| 158 | + public static void reset() { |
| 159 | + isChunked = false; |
| 160 | + } |
| 161 | + |
| 162 | + @Override |
| 163 | + public void beforeTransmission(Context.BeforeTransmission context, ExecutionAttributes executionAttributes) { |
| 164 | + isChunked = context.httpRequest().matchingHeaders("Transfer-Encoding").contains("chunked"); |
| 165 | + } |
| 166 | + } |
| 167 | + |
| 168 | + private AsyncRequestBody customAsyncRequestBodyWithoutContentLength() { |
| 169 | + return new AsyncRequestBody() { |
| 170 | + @Override |
| 171 | + public Optional<Long> contentLength() { |
| 172 | + return Optional.empty(); |
| 173 | + } |
| 174 | + |
| 175 | + @Override |
| 176 | + public void subscribe(Subscriber<? super ByteBuffer> s) { |
| 177 | + Flowable.fromPublisher(AsyncRequestBody.fromBytes("Random text".getBytes())) |
| 178 | + .subscribe(s); |
| 179 | + } |
| 180 | + }; |
| 181 | + } |
| 182 | + |
| 183 | + private static class TestContentProvider implements ContentStreamProvider { |
| 184 | + private final byte[] content; |
| 185 | + private final List<CloseTrackingInputStream> createdStreams = new ArrayList<>(); |
| 186 | + private CloseTrackingInputStream currentStream; |
| 187 | + |
| 188 | + private TestContentProvider(byte[] content) { |
| 189 | + this.content = content; |
| 190 | + } |
| 191 | + |
| 192 | + @Override |
| 193 | + public InputStream newStream() { |
| 194 | + if (currentStream != null) { |
| 195 | + invokeSafely(currentStream::close); |
| 196 | + } |
| 197 | + currentStream = new CloseTrackingInputStream(new ByteArrayInputStream(content)); |
| 198 | + createdStreams.add(currentStream); |
| 199 | + return currentStream; |
| 200 | + } |
| 201 | + |
| 202 | + List<CloseTrackingInputStream> getCreatedStreams() { |
| 203 | + return createdStreams; |
| 204 | + } |
| 205 | + } |
| 206 | + |
| 207 | + private static class CloseTrackingInputStream extends FilterInputStream { |
| 208 | + private boolean isClosed = false; |
| 209 | + |
| 210 | + CloseTrackingInputStream(InputStream in) { |
| 211 | + super(in); |
| 212 | + } |
| 213 | + |
| 214 | + @Override |
| 215 | + public void close() throws IOException { |
| 216 | + super.close(); |
| 217 | + isClosed = true; |
| 218 | + } |
| 219 | + |
| 220 | + boolean isClosed() { |
| 221 | + return isClosed; |
| 222 | + } |
| 223 | + } |
| 224 | +} |
0 commit comments