Skip to content

Commit e0b4bfc

Browse files
zoewanggzoewangg
authored
Add more tests with ByteArrayAsyncRequestBody (#4214)
1 parent 8472539 commit e0b4bfc

File tree

4 files changed

+95
-53
lines changed

4 files changed

+95
-53
lines changed

‎core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/SplittingPublisherTest.java

Lines changed: 62 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,21 @@
2020

2121
import java.io.ByteArrayInputStream;
2222
import java.io.ByteArrayOutputStream;
23+
import java.io.File;
2324
import java.io.FileInputStream;
2425
import java.io.IOException;
2526
import java.nio.ByteBuffer;
2627
import java.nio.charset.Charset;
28+
import java.nio.file.Files;
2729
import java.util.ArrayList;
2830
import java.util.List;
2931
import java.util.Optional;
32+
import java.util.UUID;
3033
import java.util.concurrent.CompletableFuture;
34+
import java.util.concurrent.ExecutionException;
35+
import java.util.concurrent.ThreadLocalRandom;
3136
import java.util.concurrent.TimeUnit;
37+
import java.util.concurrent.TimeoutException;
3238
import java.util.concurrent.atomic.AtomicInteger;
3339
import org.apache.commons.lang3.RandomStringUtils;
3440
import org.junit.jupiter.api.AfterAll;
@@ -51,11 +57,12 @@ public class SplittingPublisherTest {
5157

5258
private static final int NUM_OF_CHUNK = (int) Math.ceil(CONTENT_SIZE / (double) CHUNK_SIZE);
5359

54-
private static RandomTempFile testFile;
60+
private static File testFile;
5561

5662
@BeforeAll
5763
public static void beforeAll() throws IOException {
58-
testFile = new RandomTempFile("testfile.dat", CONTENT_SIZE);
64+
testFile = File.createTempFile("SplittingPublisherTest", UUID.randomUUID().toString());
65+
Files.write(testFile.toPath(), CONTENT);
5966
}
6067

6168
@AfterAll
@@ -65,46 +72,19 @@ public static void afterAll() throws Exception {
6572

6673
@ParameterizedTest
6774
@ValueSource(ints = {CHUNK_SIZE, CHUNK_SIZE * 2 - 1, CHUNK_SIZE * 2})
68-
void differentChunkSize_shouldSplitAsyncRequestBodyCorrectly(int upstreamByteBufferSize) throws Exception {
69-
CompletableFuture<Void> future = new CompletableFuture<>();
70-
SplittingPublisher splittingPublisher = SplittingPublisher.builder()
71-
.resultFuture(future)
72-
.asyncRequestBody(FileAsyncRequestBody.builder()
73-
.path(testFile.toPath())
74-
.chunkSizeInBytes(upstreamByteBufferSize)
75-
.build())
76-
77-
.resultFuture(future)
78-
.chunkSizeInBytes((long) CHUNK_SIZE)
79-
.maxMemoryUsageInBytes((long) CHUNK_SIZE * 4)
80-
.build();
81-
82-
List<CompletableFuture<byte[]>> futures = new ArrayList<>();
75+
void differentChunkSize_shouldSplitAsyncRequestBodyCorrectly(int chunkSize) throws Exception {
8376

84-
splittingPublisher.subscribe(requestBody -> {
85-
CompletableFuture<byte[]> baosFuture = new CompletableFuture<>();
86-
BaosSubscriber subscriber = new BaosSubscriber(baosFuture);
87-
futures.add(baosFuture);
88-
requestBody.subscribe(subscriber);
89-
}).get(5, TimeUnit.SECONDS);
90-
91-
assertThat(futures.size()).isEqualTo(NUM_OF_CHUNK);
77+
FileAsyncRequestBody fileAsyncRequestBody = FileAsyncRequestBody.builder()
78+
.path(testFile.toPath())
79+
.chunkSizeInBytes(chunkSize)
80+
.build();
81+
verifySplitContent(fileAsyncRequestBody, chunkSize);
82+
}
9283

93-
for (int i = 0; i < futures.size(); i++) {
94-
try (FileInputStream fileInputStream = new FileInputStream(testFile)) {
95-
byte[] expected;
96-
if (i == futures.size() - 1) {
97-
expected = new byte[1];
98-
} else {
99-
expected = new byte[5];
100-
}
101-
fileInputStream.skip(i * 5);
102-
fileInputStream.read(expected);
103-
byte[] actualBytes = futures.get(i).join();
104-
assertThat(actualBytes).isEqualTo(expected);
105-
};
106-
}
107-
assertThat(future).isCompleted();
84+
@ParameterizedTest
85+
@ValueSource(ints = {CHUNK_SIZE, CHUNK_SIZE * 2 - 1, CHUNK_SIZE * 2})
86+
void differentChunkSize_byteArrayShouldSplitAsyncRequestBodyCorrectly(int chunkSize) throws Exception {
87+
verifySplitContent(AsyncRequestBody.fromBytes(CONTENT), chunkSize);
10888
}
10989

11090

@@ -115,7 +95,7 @@ void cancelFuture_shouldCancelUpstream() throws IOException {
11595
SplittingPublisher splittingPublisher = SplittingPublisher.builder()
11696
.resultFuture(future)
11797
.asyncRequestBody(asyncRequestBody)
118-
.chunkSizeInBytes((long) CHUNK_SIZE)
98+
.chunkSizeInBytes(CHUNK_SIZE)
11999
.maxMemoryUsageInBytes(10L)
120100
.build();
121101

@@ -139,7 +119,7 @@ public Optional<Long> contentLength() {
139119
SplittingPublisher splittingPublisher = SplittingPublisher.builder()
140120
.resultFuture(future)
141121
.asyncRequestBody(asyncRequestBody)
142-
.chunkSizeInBytes((long) CHUNK_SIZE)
122+
.chunkSizeInBytes(CHUNK_SIZE)
143123
.maxMemoryUsageInBytes(10L)
144124
.build();
145125

@@ -177,6 +157,46 @@ public Optional<Long> contentLength() {
177157

178158
}
179159

160+
161+
private static void verifySplitContent(AsyncRequestBody asyncRequestBody, int chunkSize) throws Exception {
162+
CompletableFuture<Void> future = new CompletableFuture<>();
163+
SplittingPublisher splittingPublisher = SplittingPublisher.builder()
164+
.resultFuture(future)
165+
.asyncRequestBody(asyncRequestBody)
166+
.resultFuture(future)
167+
.chunkSizeInBytes(chunkSize)
168+
.maxMemoryUsageInBytes((long) chunkSize * 4)
169+
.build();
170+
171+
List<CompletableFuture<byte[]>> futures = new ArrayList<>();
172+
173+
splittingPublisher.subscribe(requestBody -> {
174+
CompletableFuture<byte[]> baosFuture = new CompletableFuture<>();
175+
BaosSubscriber subscriber = new BaosSubscriber(baosFuture);
176+
futures.add(baosFuture);
177+
requestBody.subscribe(subscriber);
178+
}).get(5, TimeUnit.SECONDS);
179+
180+
assertThat(futures.size()).isEqualTo((int) Math.ceil(CONTENT_SIZE / (double) chunkSize));
181+
182+
for (int i = 0; i < futures.size(); i++) {
183+
try (FileInputStream fileInputStream = new FileInputStream(testFile)) {
184+
byte[] expected;
185+
if (i == futures.size() - 1) {
186+
int lastChunk = CONTENT_SIZE % chunkSize == 0 ? chunkSize : (CONTENT_SIZE % chunkSize);
187+
expected = new byte[lastChunk];
188+
} else {
189+
expected = new byte[chunkSize];
190+
}
191+
fileInputStream.skip(i * chunkSize);
192+
fileInputStream.read(expected);
193+
byte[] actualBytes = futures.get(i).join();
194+
assertThat(actualBytes).isEqualTo(expected);
195+
};
196+
}
197+
assertThat(future).isCompleted();
198+
}
199+
180200
private static class TestAsyncRequestBody implements AsyncRequestBody {
181201
private volatile boolean cancelled;
182202
private volatile boolean isDone;

‎services/s3/src/it/java/software/amazon/awssdk/services/s3/crt/S3CrossRegionCrtIntegrationTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
import static org.assertj.core.api.Assertions.assertThat;
1919
import static org.assertj.core.api.Assertions.assertThatThrownBy;
20-
import static software.amazon.awssdk.services.s3.crt.S3CrtClientCopyIntegrationTest.randomBytes;
20+
import static software.amazon.awssdk.services.s3.multipart.S3ClientMultiPartCopyIntegrationTest.randomBytes;
2121
import static software.amazon.awssdk.services.s3.utils.ChecksumUtils.computeCheckSum;
2222
import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
2323

‎services/s3/src/it/java/software/amazon/awssdk/services/s3/multipart/S3ClientMultiPartCopyIntegrationTest.java

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -119,14 +119,12 @@ void copy_ssecServerSideEncryption_shouldSucceed(S3AsyncClient s3AsyncClient) {
119119
String newB64Key = Base64.getEncoder().encodeToString(newSecretKey);
120120
String newB64KeyMd5 = Md5Utils.md5AsBase64(newSecretKey);
121121

122-
// MPU S3 client gets stuck
123-
// TODO: change back to s3AsyncClient once the issue is fixed in MPU S3 client
124-
s3Async.putObject(r -> r.bucket(BUCKET)
125-
.key(ORIGINAL_OBJ)
126-
.sseCustomerKey(b64Key)
127-
.sseCustomerAlgorithm(AES256.name())
128-
.sseCustomerKeyMD5(b64KeyMd5),
129-
AsyncRequestBody.fromBytes(originalContent)).join();
122+
s3AsyncClient.putObject(r -> r.bucket(BUCKET)
123+
.key(ORIGINAL_OBJ)
124+
.sseCustomerKey(b64Key)
125+
.sseCustomerAlgorithm(AES256.name())
126+
.sseCustomerKeyMD5(b64KeyMd5),
127+
AsyncRequestBody.fromBytes(originalContent)).join();
130128

131129
CompletableFuture<CopyObjectResponse> future = s3AsyncClient.copyObject(c -> c
132130
.sourceBucket(BUCKET)

‎services/s3/src/it/java/software/amazon/awssdk/services/s3/multipart/S3MultipartClientPutObjectIntegrationTest.java

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,13 @@
2020
import static org.assertj.core.api.Assertions.assertThat;
2121
import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
2222

23+
import java.io.ByteArrayInputStream;
24+
import java.io.File;
25+
import java.nio.charset.Charset;
2326
import java.nio.file.Files;
27+
import java.util.UUID;
28+
import java.util.concurrent.ThreadLocalRandom;
29+
import org.apache.commons.lang3.RandomStringUtils;
2430
import org.assertj.core.api.Assertions;
2531
import org.junit.jupiter.api.AfterAll;
2632
import org.junit.jupiter.api.BeforeAll;
@@ -42,15 +48,18 @@ public class S3MultipartClientPutObjectIntegrationTest extends S3IntegrationTest
4248
private static final String TEST_KEY = "testfile.dat";
4349
private static final int OBJ_SIZE = 19 * 1024 * 1024;
4450

45-
private static RandomTempFile testFile;
51+
private static File testFile;
4652
private static S3AsyncClient mpuS3Client;
4753

4854
@BeforeAll
4955
public static void setup() throws Exception {
5056
S3IntegrationTestBase.setUp();
5157
S3IntegrationTestBase.createBucket(TEST_BUCKET);
58+
byte[] CONTENT =
59+
RandomStringUtils.randomAscii(OBJ_SIZE).getBytes(Charset.defaultCharset());
5260

53-
testFile = new RandomTempFile(TEST_KEY, OBJ_SIZE);
61+
testFile = File.createTempFile("SplittingPublisherTest", UUID.randomUUID().toString());
62+
Files.write(testFile.toPath(), CONTENT);
5463
mpuS3Client = new MultipartS3AsyncClient(s3Async);
5564
}
5665

@@ -75,4 +84,19 @@ void putObject_fileRequestBody_objectSentCorrectly() throws Exception {
7584
assertThat(ChecksumUtils.computeCheckSum(objContent)).isEqualTo(expectedSum);
7685
}
7786

87+
@Test
88+
@Timeout(value = 30, unit = SECONDS)
89+
void putObject_byteAsyncRequestBody_objectSentCorrectly() throws Exception {
90+
byte[] bytes = RandomStringUtils.randomAscii(OBJ_SIZE).getBytes(Charset.defaultCharset());
91+
AsyncRequestBody body = AsyncRequestBody.fromBytes(bytes);
92+
mpuS3Client.putObject(r -> r.bucket(TEST_BUCKET).key(TEST_KEY), body).join();
93+
94+
ResponseInputStream<GetObjectResponse> objContent = S3IntegrationTestBase.s3.getObject(r -> r.bucket(TEST_BUCKET).key(TEST_KEY),
95+
ResponseTransformer.toInputStream());
96+
97+
assertThat(objContent.response().contentLength()).isEqualTo(OBJ_SIZE);
98+
byte[] expectedSum = ChecksumUtils.computeCheckSum(new ByteArrayInputStream(bytes));
99+
assertThat(ChecksumUtils.computeCheckSum(objContent)).isEqualTo(expectedSum);
100+
}
101+
78102
}

0 commit comments

Comments
 (0)