* {@link #subscribe(Subscriber)} should be implemented to tie this publisher to a subscriber. Ideally each call to subscribe - * should reproduce the content (i.e if you are reading from a file each subscribe call should produce a {@link - * org.reactivestreams.Subscription} that reads the file fully). This allows for automatic retries to be performed in the SDK. If - * the content is not reproducible, an exception may be thrown from any subsequent {@link #subscribe(Subscriber)} calls. + * should reproduce the content (i.e if you are reading from a file each subscribe call should produce a + * {@link org.reactivestreams.Subscription} that reads the file fully). This allows for automatic retries to be performed in the + * SDK. If the content is not reproducible, an exception may be thrown from any subsequent {@link #subscribe(Subscriber)} calls. *
* *- * It is important to only send the number of chunks that the subscriber requests to avoid out of memory situations. - * The subscriber does it's own buffering so it's usually not needed to buffer in the publisher. Additional permits - * for chunks will be notified via the {@link org.reactivestreams.Subscription#request(long)} method. + * It is important to only send the number of chunks that the subscriber requests to avoid out of memory situations. The + * subscriber does it's own buffering so it's usually not needed to buffer in the publisher. Additional permits for chunks will be + * notified via the {@link org.reactivestreams.Subscription#request(long)} method. *
* * @see FileAsyncRequestBody - * @see ByteArrayAsyncRequestBody + * @see ByteBuffersAsyncRequestBody */ @SdkPublicApi public interface AsyncRequestBody extends SdkPublisherAs the method name implies, this is unsafe. Use {@link #fromBytes(byte[])} unless you're sure you know the risks. + * + * @param bytes The bytes to send to the service. + * @return AsyncRequestBody instance. + */ + static AsyncRequestBody fromBytesUnsafe(byte[] bytes) { + return ByteBuffersAsyncRequestBody.from(bytes); + } + + /** + * Creates an {@link AsyncRequestBody} from a {@link ByteBuffer}. This will copy the contents of the {@link ByteBuffer} to + * prevent modifications to the provided {@link ByteBuffer} from being reflected in the {@link AsyncRequestBody}. + *
+ * NOTE: This method ignores the current read position. Use {@link #fromRemainingByteBuffer(ByteBuffer)} if you need + * it to copy only the remaining readable bytes. * * @param byteBuffer ByteBuffer to send to the service. * @return AsyncRequestBody instance. */ static AsyncRequestBody fromByteBuffer(ByteBuffer byteBuffer) { - return fromBytes(BinaryUtils.copyAllBytesFrom(byteBuffer)); + ByteBuffer immutableCopy = BinaryUtils.immutableCopyOf(byteBuffer); + immutableCopy.rewind(); + return ByteBuffersAsyncRequestBody.of((long) immutableCopy.remaining(), immutableCopy); + } + + /** + * Creates an {@link AsyncRequestBody} from the remaining readable bytes from a {@link ByteBuffer}. This will copy the + * remaining contents of the {@link ByteBuffer} to prevent modifications to the provided {@link ByteBuffer} from being + * reflected in the {@link AsyncRequestBody}. + *
Unlike {@link #fromByteBuffer(ByteBuffer)}, this method respects the current read position of the buffer and reads + * only the remaining bytes. + * + * @param byteBuffer ByteBuffer to send to the service. + * @return AsyncRequestBody instance. + */ + static AsyncRequestBody fromRemainingByteBuffer(ByteBuffer byteBuffer) { + ByteBuffer immutableCopy = BinaryUtils.immutableCopyOfRemaining(byteBuffer); + return ByteBuffersAsyncRequestBody.of((long) immutableCopy.remaining(), immutableCopy); + } + + /** + * Creates an {@link AsyncRequestBody} from a {@link ByteBuffer} without copying the contents of the + * {@link ByteBuffer}. This introduces concurrency risks, allowing the caller to modify the {@link ByteBuffer} stored in this + * {@code AsyncRequestBody} implementation. + *
+ * NOTE: This method ignores the current read position. Use {@link #fromRemainingByteBufferUnsafe(ByteBuffer)} if you + * need it to copy only the remaining readable bytes. + * + *
As the method name implies, this is unsafe. Use {@link #fromByteBuffer(ByteBuffer)}} unless you're sure you know the + * risks. + * + * @param byteBuffer ByteBuffer to send to the service. + * @return AsyncRequestBody instance. + */ + static AsyncRequestBody fromByteBufferUnsafe(ByteBuffer byteBuffer) { + ByteBuffer readOnlyBuffer = byteBuffer.asReadOnlyBuffer(); + readOnlyBuffer.rewind(); + return ByteBuffersAsyncRequestBody.of((long) readOnlyBuffer.remaining(), readOnlyBuffer); + } + + /** + * Creates an {@link AsyncRequestBody} from a {@link ByteBuffer} without copying the contents of the + * {@link ByteBuffer}. This introduces concurrency risks, allowing the caller to modify the {@link ByteBuffer} stored in this + * {@code AsyncRequestBody} implementation. + *
Unlike {@link #fromByteBufferUnsafe(ByteBuffer)}, this method respects the current read position of + * the buffer and reads only the remaining bytes. + * + *
As the method name implies, this is unsafe. Use {@link #fromByteBuffer(ByteBuffer)}} unless you're sure you know the + * risks. + * + * @param byteBuffer ByteBuffer to send to the service. + * @return AsyncRequestBody instance. + */ + static AsyncRequestBody fromRemainingByteBufferUnsafe(ByteBuffer byteBuffer) { + ByteBuffer readOnlyBuffer = byteBuffer.asReadOnlyBuffer(); + return ByteBuffersAsyncRequestBody.of((long) readOnlyBuffer.remaining(), readOnlyBuffer); + } + + /** + * Creates an {@link AsyncRequestBody} from a {@link ByteBuffer} array. This will copy the contents of each {@link ByteBuffer} + * to prevent modifications to any provided {@link ByteBuffer} from being reflected in the {@link AsyncRequestBody}. + *
+ * NOTE: This method ignores the current read position of each {@link ByteBuffer}. Use + * {@link #fromRemainingByteBuffers(ByteBuffer...)} if you need it to copy only the remaining readable bytes. + * + * @param byteBuffers ByteBuffer array to send to the service. + * @return AsyncRequestBody instance. + */ + static AsyncRequestBody fromByteBuffers(ByteBuffer... byteBuffers) { + ByteBuffer[] immutableCopy = Arrays.stream(byteBuffers) + .map(BinaryUtils::immutableCopyOf) + .peek(ByteBuffer::rewind) + .toArray(ByteBuffer[]::new); + return ByteBuffersAsyncRequestBody.of(immutableCopy); + } + + /** + * Creates an {@link AsyncRequestBody} from a {@link ByteBuffer} array. This will copy the remaining contents of each + * {@link ByteBuffer} to prevent modifications to any provided {@link ByteBuffer} from being reflected in the + * {@link AsyncRequestBody}. + *
Unlike {@link #fromByteBufferUnsafe(ByteBuffer)}, + * this method respects the current read position of each buffer and reads only the remaining bytes. + * + * @param byteBuffers ByteBuffer array to send to the service. + * @return AsyncRequestBody instance. + */ + static AsyncRequestBody fromRemainingByteBuffers(ByteBuffer... byteBuffers) { + ByteBuffer[] immutableCopy = Arrays.stream(byteBuffers) + .map(BinaryUtils::immutableCopyOfRemaining) + .peek(ByteBuffer::rewind) + .toArray(ByteBuffer[]::new); + return ByteBuffersAsyncRequestBody.of(immutableCopy); + } + + /** + * Creates an {@link AsyncRequestBody} from a {@link ByteBuffer} array without copying the contents of each + * {@link ByteBuffer}. This introduces concurrency risks, allowing the caller to modify any {@link ByteBuffer} stored in this + * {@code AsyncRequestBody} implementation. + *
+ * NOTE: This method ignores the current read position of each {@link ByteBuffer}. Use + * {@link #fromRemainingByteBuffers(ByteBuffer...)} if you need it to copy only the remaining readable bytes. + * + *
As the method name implies, this is unsafe. Use {@link #fromByteBuffers(ByteBuffer...)} unless you're sure you know the + * risks. + * + * @param byteBuffers ByteBuffer array to send to the service. + * @return AsyncRequestBody instance. + */ + static AsyncRequestBody fromByteBuffersUnsafe(ByteBuffer... byteBuffers) { + ByteBuffer[] readOnlyBuffers = Arrays.stream(byteBuffers) + .map(ByteBuffer::asReadOnlyBuffer) + .peek(ByteBuffer::rewind) + .toArray(ByteBuffer[]::new); + return ByteBuffersAsyncRequestBody.of(readOnlyBuffers); + } + + /** + * Creates an {@link AsyncRequestBody} from a {@link ByteBuffer} array without copying the contents of each + * {@link ByteBuffer}. This introduces concurrency risks, allowing the caller to modify any {@link ByteBuffer} stored in this + * {@code AsyncRequestBody} implementation. + *
Unlike {@link #fromByteBuffersUnsafe(ByteBuffer...)}, + * this method respects the current read position of each buffer and reads only the remaining bytes. + * + *
As the method name implies, this is unsafe. Use {@link #fromByteBuffers(ByteBuffer...)} unless you're sure you know the + * risks. + * + * @param byteBuffers ByteBuffer array to send to the service. + * @return AsyncRequestBody instance. + */ + static AsyncRequestBody fromRemainingByteBuffersUnsafe(ByteBuffer... byteBuffers) { + ByteBuffer[] readOnlyBuffers = Arrays.stream(byteBuffers) + .map(ByteBuffer::asReadOnlyBuffer) + .toArray(ByteBuffer[]::new); + return ByteBuffersAsyncRequestBody.of(readOnlyBuffers); } /** - * Creates a {@link AsyncRequestBody} from a {@link InputStream}. + * Creates an {@link AsyncRequestBody} from an {@link InputStream}. * *
An {@link ExecutorService} is required in order to perform the blocking data reads, to prevent blocking the
* non-blocking event loop threads owned by the SDK.
@@ -242,7 +395,7 @@ static BlockingOutputStreamAsyncRequestBody forBlockingOutputStream(Long content
}
/**
- * Creates a {@link AsyncRequestBody} with no content.
+ * Creates an {@link AsyncRequestBody} with no content.
*
* @return AsyncRequestBody instance.
*/
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ByteArrayAsyncRequestBody.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ByteArrayAsyncRequestBody.java
deleted file mode 100644
index 29205479b798..000000000000
--- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ByteArrayAsyncRequestBody.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * 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.async;
-
-import java.nio.ByteBuffer;
-import java.util.Optional;
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
-import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.core.async.AsyncRequestBody;
-import software.amazon.awssdk.utils.Logger;
-
-/**
- * An implementation of {@link AsyncRequestBody} for providing data from memory. This is created using static
- * methods on {@link AsyncRequestBody}
- *
- * @see AsyncRequestBody#fromBytes(byte[])
- * @see AsyncRequestBody#fromByteBuffer(ByteBuffer)
- * @see AsyncRequestBody#fromString(String)
- */
-@SdkInternalApi
-public final class ByteArrayAsyncRequestBody implements AsyncRequestBody {
- private static final Logger log = Logger.loggerFor(ByteArrayAsyncRequestBody.class);
-
- private final byte[] bytes;
-
- private final String mimetype;
-
- public ByteArrayAsyncRequestBody(byte[] bytes, String mimetype) {
- this.bytes = bytes.clone();
- this.mimetype = mimetype;
- }
-
- @Override
- public Optional
+ * The new buffer's position will be set to the position of the given {@code ByteBuffer}, but the mark if defined will be
+ * ignored.
+ *
+ * NOTE: this method intentionally converts direct buffers to non-direct though there is no guarantee this will always
+ * be the case, if this is required see {@link #toNonDirectBuffer(ByteBuffer)}
+ *
+ * @param bb the source {@code ByteBuffer} to copy.
+ * @return a read only {@code ByteBuffer}.
+ */
+ public static ByteBuffer immutableCopyOf(ByteBuffer bb) {
+ if (bb == null) {
+ return null;
+ }
+ int sourceBufferPosition = bb.position();
+ ByteBuffer readOnlyCopy = bb.asReadOnlyBuffer();
+ readOnlyCopy.rewind();
+ ByteBuffer cloned = ByteBuffer.allocate(readOnlyCopy.capacity())
+ .put(readOnlyCopy);
+ cloned.position(sourceBufferPosition);
+ return cloned.asReadOnlyBuffer();
+ }
+
+ /**
+ * Returns an immutable copy of the remaining bytes of the given {@code ByteBuffer}.
+ *
+ * NOTE: this method intentionally converts direct buffers to non-direct though there is no guarantee this will always
+ * be the case, if this is required see {@link #toNonDirectBuffer(ByteBuffer)}
+ *
+ * @param bb the source {@code ByteBuffer} to copy.
+ * @return a read only {@code ByteBuffer}.
+ */
+ public static ByteBuffer immutableCopyOfRemaining(ByteBuffer bb) {
+ if (bb == null) {
+ return null;
+ }
+ ByteBuffer readOnlyCopy = bb.asReadOnlyBuffer();
+ ByteBuffer cloned = ByteBuffer.allocate(readOnlyCopy.remaining())
+ .put(readOnlyCopy);
+ cloned.flip();
+ return cloned.asReadOnlyBuffer();
+ }
+
+ /**
+ * Returns a copy of the given {@code DirectByteBuffer} from its current position as a non-direct {@code HeapByteBuffer}
+ *
+ * The new buffer's position will be set to the position of the given {@code ByteBuffer}, but the mark if defined will be
+ * ignored.
+ *
+ * @param bb the source {@code ByteBuffer} to copy.
+ * @return {@code ByteBuffer}.
+ */
+ public static ByteBuffer toNonDirectBuffer(ByteBuffer bb) {
+ if (bb == null) {
+ return null;
+ }
+ if (!bb.isDirect()) {
+ throw new IllegalArgumentException("Provided ByteBuffer is already non-direct");
+ }
+ int sourceBufferPosition = bb.position();
+ ByteBuffer readOnlyCopy = bb.asReadOnlyBuffer();
+ readOnlyCopy.rewind();
+ ByteBuffer cloned = ByteBuffer.allocate(bb.capacity())
+ .put(readOnlyCopy);
+ cloned.rewind();
+ cloned.position(sourceBufferPosition);
+ if (bb.isReadOnly()) {
+ return cloned.asReadOnlyBuffer();
+ }
+ return cloned;
+ }
+
/**
* Returns a copy of all the bytes from the given ByteBuffer,
* from the beginning to the buffer's limit; or null if the input is null.
diff --git a/utils/src/test/java/software/amazon/awssdk/utils/BinaryUtilsTest.java b/utils/src/test/java/software/amazon/awssdk/utils/BinaryUtilsTest.java
index 5f255d347adc..4e416ea9e3b6 100644
--- a/utils/src/test/java/software/amazon/awssdk/utils/BinaryUtilsTest.java
+++ b/utils/src/test/java/software/amazon/awssdk/utils/BinaryUtilsTest.java
@@ -16,9 +16,11 @@
package software.amazon.awssdk.utils;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.ByteBuffer;
@@ -32,13 +34,11 @@ public class BinaryUtilsTest {
public void testHex() {
{
String hex = BinaryUtils.toHex(new byte[] {0});
- System.out.println(hex);
String hex2 = Base16Lower.encodeAsString(new byte[] {0});
assertEquals(hex, hex2);
}
{
String hex = BinaryUtils.toHex(new byte[] {-1});
- System.out.println(hex);
String hex2 = Base16Lower.encodeAsString(new byte[] {-1});
assertEquals(hex, hex2);
}
@@ -169,7 +169,7 @@ public void testCopyRemainingBytesFrom_nullBuffer() {
@Test
public void testCopyRemainingBytesFrom_noRemainingBytes() {
ByteBuffer bb = ByteBuffer.allocate(1);
- bb.put(new byte[]{1});
+ bb.put(new byte[] {1});
bb.flip();
bb.get();
@@ -180,7 +180,7 @@ public void testCopyRemainingBytesFrom_noRemainingBytes() {
@Test
public void testCopyRemainingBytesFrom_fullBuffer() {
ByteBuffer bb = ByteBuffer.allocate(4);
- bb.put(new byte[]{1, 2, 3, 4});
+ bb.put(new byte[] {1, 2, 3, 4});
bb.flip();
byte[] copy = BinaryUtils.copyRemainingBytesFrom(bb);
@@ -191,7 +191,7 @@ public void testCopyRemainingBytesFrom_fullBuffer() {
@Test
public void testCopyRemainingBytesFrom_partiallyReadBuffer() {
ByteBuffer bb = ByteBuffer.allocate(4);
- bb.put(new byte[]{1, 2, 3, 4});
+ bb.put(new byte[] {1, 2, 3, 4});
bb.flip();
bb.get();
@@ -201,4 +201,137 @@ public void testCopyRemainingBytesFrom_partiallyReadBuffer() {
assertThat(bb).isEqualTo(ByteBuffer.wrap(copy));
assertThat(copy).hasSize(2);
}
+
+ @Test
+ public void testImmutableCopyOfByteBuffer() {
+ ByteBuffer sourceBuffer = ByteBuffer.allocate(4);
+ byte[] originalBytesInSource = {1, 2, 3, 4};
+ sourceBuffer.put(originalBytesInSource);
+ sourceBuffer.flip();
+
+ ByteBuffer immutableCopy = BinaryUtils.immutableCopyOf(sourceBuffer);
+
+ byte[] bytesInSourceAfterCopy = {-1, -2, -3, -4};
+ sourceBuffer.put(bytesInSourceAfterCopy);
+ sourceBuffer.flip();
+
+ assertTrue(immutableCopy.isReadOnly());
+ byte[] fromImmutableCopy = new byte[originalBytesInSource.length];
+ immutableCopy.get(fromImmutableCopy);
+ assertArrayEquals(originalBytesInSource, fromImmutableCopy);
+
+ assertEquals(0, sourceBuffer.position());
+ byte[] fromSource = new byte[bytesInSourceAfterCopy.length];
+ sourceBuffer.get(fromSource);
+ assertArrayEquals(bytesInSourceAfterCopy, fromSource);
+ }
+
+ @Test
+ public void testImmutableCopyOfByteBuffer_nullBuffer() {
+ assertNull(BinaryUtils.immutableCopyOf(null));
+ }
+
+ @Test
+ public void testImmutableCopyOfByteBuffer_partiallyReadBuffer() {
+ ByteBuffer sourceBuffer = ByteBuffer.allocate(4);
+ byte[] bytes = {1, 2, 3, 4};
+ sourceBuffer.put(bytes);
+ sourceBuffer.position(2);
+
+ ByteBuffer immutableCopy = BinaryUtils.immutableCopyOf(sourceBuffer);
+
+ assertEquals(sourceBuffer.position(), immutableCopy.position());
+ immutableCopy.rewind();
+ byte[] fromImmutableCopy = new byte[bytes.length];
+ immutableCopy.get(fromImmutableCopy);
+ assertArrayEquals(bytes, fromImmutableCopy);
+ }
+
+ @Test
+ public void testImmutableCopyOfRemainingByteBuffer() {
+ ByteBuffer sourceBuffer = ByteBuffer.allocate(4);
+ byte[] originalBytesInSource = {1, 2, 3, 4};
+ sourceBuffer.put(originalBytesInSource);
+ sourceBuffer.flip();
+
+ ByteBuffer immutableCopy = BinaryUtils.immutableCopyOfRemaining(sourceBuffer);
+
+ byte[] bytesInSourceAfterCopy = {-1, -2, -3, -4};
+ sourceBuffer.put(bytesInSourceAfterCopy);
+ sourceBuffer.flip();
+
+ assertTrue(immutableCopy.isReadOnly());
+ byte[] fromImmutableCopy = new byte[originalBytesInSource.length];
+ immutableCopy.get(fromImmutableCopy);
+ assertArrayEquals(originalBytesInSource, fromImmutableCopy);
+
+ assertEquals(0, sourceBuffer.position());
+ byte[] fromSource = new byte[bytesInSourceAfterCopy.length];
+ sourceBuffer.get(fromSource);
+ assertArrayEquals(bytesInSourceAfterCopy, fromSource);
+ }
+
+ @Test
+ public void testImmutableCopyOfByteBufferRemaining_nullBuffer() {
+ assertNull(BinaryUtils.immutableCopyOfRemaining(null));
+ }
+
+ @Test
+ public void testImmutableCopyOfByteBufferRemaining_partiallyReadBuffer() {
+ ByteBuffer sourceBuffer = ByteBuffer.allocate(4);
+ byte[] bytes = {1, 2, 3, 4};
+ sourceBuffer.put(bytes);
+ sourceBuffer.position(2);
+
+ ByteBuffer immutableCopy = BinaryUtils.immutableCopyOfRemaining(sourceBuffer);
+
+ assertEquals(2, immutableCopy.capacity());
+ assertEquals(2, immutableCopy.remaining());
+ assertEquals(0, immutableCopy.position());
+ assertEquals((byte) 3, immutableCopy.get());
+ assertEquals((byte) 4, immutableCopy.get());
+ }
+
+ @Test
+ public void testToNonDirectBuffer() {
+ ByteBuffer bb = ByteBuffer.allocateDirect(4);
+ byte[] expected = {1, 2, 3, 4};
+ bb.put(expected);
+ bb.flip();
+
+ ByteBuffer nonDirectBuffer = BinaryUtils.toNonDirectBuffer(bb);
+
+ assertFalse(nonDirectBuffer.isDirect());
+ byte[] bytes = new byte[expected.length];
+ nonDirectBuffer.get(bytes);
+ assertArrayEquals(expected, bytes);
+ }
+
+ @Test
+ public void testToNonDirectBuffer_nullBuffer() {
+ assertNull(BinaryUtils.toNonDirectBuffer(null));
+ }
+
+ @Test
+ public void testToNonDirectBuffer_partiallyReadBuffer() {
+ ByteBuffer sourceBuffer = ByteBuffer.allocateDirect(4);
+ byte[] bytes = {1, 2, 3, 4};
+ sourceBuffer.put(bytes);
+ sourceBuffer.position(2);
+
+ ByteBuffer nonDirectBuffer = BinaryUtils.toNonDirectBuffer(sourceBuffer);
+
+ assertEquals(sourceBuffer.position(), nonDirectBuffer.position());
+ nonDirectBuffer.rewind();
+ byte[] fromNonDirectBuffer = new byte[bytes.length];
+ nonDirectBuffer.get(fromNonDirectBuffer);
+ assertArrayEquals(bytes, fromNonDirectBuffer);
+ }
+
+ @Test
+ public void testToNonDirectBuffer_nonDirectBuffer() {
+ ByteBuffer nonDirectBuffer = ByteBuffer.allocate(0);
+ assertThrows(IllegalArgumentException.class, () -> BinaryUtils.toNonDirectBuffer(nonDirectBuffer));
+ }
+
}