diff --git a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageConstants.java b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageConstants.java index 440c4521f593..859ac4c972b6 100644 --- a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageConstants.java +++ b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageConstants.java @@ -26,4 +26,19 @@ public final class StructuredMessageConstants { * The length of the CRC64 checksum. */ public static final int CRC64_LENGTH = 8; + + /** + * The default length of segments for version 1. + */ + public static final int V1_DEFAULT_SEGMENT_CONTENT_LENGTH = 4 * 1024 * 1024; // 4 MiB + + /** + * The maximum amount of data to encode at once. + */ + public static final int STATIC_MAXIMUM_ENCODED_DATA_LENGTH = 4 * 1024 * 1024; // 4 MiB + + /** + * The header name for the CRC64 checksum. + */ + public static final String STRUCTURED_BODY_TYPE_VALUE = "XSM/1.0; properties=crc64"; } diff --git a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageEncoder.java b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageEncoder.java index 24f80a591948..c4951b0d8e81 100644 --- a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageEncoder.java +++ b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/implementation/structuredmessage/StructuredMessageEncoder.java @@ -4,7 +4,6 @@ package com.azure.storage.common.implementation.structuredmessage; import com.azure.core.util.logging.ClientLogger; -import com.azure.storage.common.implementation.BufferStagingArea; import com.azure.storage.common.implementation.StorageCrc64Calculator; import com.azure.storage.common.implementation.StorageImplUtils; import reactor.core.publisher.Flux; @@ -109,7 +108,9 @@ private byte[] generateMessageHeader() { return buffer.array(); } - private byte[] generateSegmentHeader(int segmentContentSize) { + private byte[] generateSegmentHeader() { + int segmentContentSize = Math.min(segmentSize, contentLength - currentContentOffset); + // 2 byte number, 8 byte size ByteBuffer buffer = ByteBuffer.allocate(getSegmentHeaderLength()).order(ByteOrder.LITTLE_ENDIAN); buffer.putShort((short) currentSegmentNumber); buffer.putLong(segmentContentSize); @@ -119,9 +120,6 @@ private byte[] generateSegmentHeader(int segmentContentSize) { /** * Encodes the given buffer into a structured message format as a stream of ByteBuffers. - * The encoder maintains mutable state and is designed for single, sequential subscription only. - * Callers should pre-chunk input buffers to appropriate sizes (e.g., using {@link BufferStagingArea}) to - * control memory usage. * * @param unencodedBuffer The buffer to be encoded. * @return A Flux of encoded ByteBuffers. @@ -146,6 +144,7 @@ public Flux encode(ByteBuffer unencodedBuffer) { return Flux.empty(); } + // create a list of buffers to store the encoded message List buffers = new ArrayList<>(); // if we are at the beginning of the message, encode message header @@ -157,9 +156,7 @@ public Flux encode(ByteBuffer unencodedBuffer) { // if we are at the beginning of a segment's content, encode segment header if (currentSegmentOffset == 0) { incrementCurrentSegment(); - // Calculate actual segment size based on remaining content - int actualSegmentSize = Math.min(segmentSize, contentLength - currentContentOffset); - buffers.add(ByteBuffer.wrap(generateSegmentHeader(actualSegmentSize))); + buffers.add(ByteBuffer.wrap(generateSegmentHeader())); } buffers.add(encodeSegmentContent(unencodedBuffer)); diff --git a/sdk/storage/azure-storage-common/src/test/java/com/azure/storage/common/implementation/structuredmessage/MessageEncoderTests.java b/sdk/storage/azure-storage-common/src/test/java/com/azure/storage/common/implementation/structuredmessage/MessageEncoderTests.java index aba5ebf2d706..426ed5454abd 100644 --- a/sdk/storage/azure-storage-common/src/test/java/com/azure/storage/common/implementation/structuredmessage/MessageEncoderTests.java +++ b/sdk/storage/azure-storage-common/src/test/java/com/azure/storage/common/implementation/structuredmessage/MessageEncoderTests.java @@ -3,27 +3,28 @@ package com.azure.storage.common.implementation.structuredmessage; +import com.azure.core.util.FluxUtil; import com.azure.storage.common.implementation.StorageCrc64Calculator; -import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; -import reactor.test.StepVerifier; +import org.junit.jupiter.api.Disabled; +import reactor.core.publisher.Flux; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.Arrays; -import java.util.Objects; import java.util.concurrent.ThreadLocalRandom; import java.util.stream.Stream; -import static com.azure.core.util.FluxUtil.collectBytesInByteBufferStream; +import static com.azure.storage.common.implementation.structuredmessage.StructuredMessageConstants.V1_DEFAULT_SEGMENT_CONTENT_LENGTH; import static com.azure.storage.common.implementation.structuredmessage.StructuredMessageConstants.CRC64_LENGTH; import static com.azure.storage.common.implementation.structuredmessage.StructuredMessageConstants.V1_HEADER_LENGTH; import static com.azure.storage.common.implementation.structuredmessage.StructuredMessageConstants.V1_SEGMENT_HEADER_LENGTH; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -159,10 +160,11 @@ public void readAll(int size, int segmentSize, StructuredMessageFlags flags) thr StructuredMessageEncoder structuredMessageEncoder = new StructuredMessageEncoder(size, segmentSize, flags); - byte[] actual = collectBytesInByteBufferStream(structuredMessageEncoder.encode(unencodedBuffer)).block(); + byte[] actual + = FluxUtil.collectBytesInByteBufferStream(structuredMessageEncoder.encode(unencodedBuffer)).block(); byte[] expected = buildStructuredMessage(unencodedBuffer, segmentSize, flags).array(); - Assertions.assertArrayEquals(expected, actual); + assertArrayEquals(expected, actual); } private static Stream readMultipleSupplier() { @@ -193,42 +195,40 @@ public void readMultiple(int segmentSize, StructuredMessageFlags flags) throws I byte[] expected = buildStructuredMessage(allWrappedData, segmentSize, flags).array(); - ByteArrayOutputStream allActualData = new ByteArrayOutputStream(); - allActualData.write(Objects - .requireNonNull(collectBytesInByteBufferStream(structuredMessageEncoder.encode(wrappedData1)).block())); - allActualData.write(Objects - .requireNonNull(collectBytesInByteBufferStream(structuredMessageEncoder.encode(wrappedData2)).block())); - allActualData.write(Objects - .requireNonNull(collectBytesInByteBufferStream(structuredMessageEncoder.encode(wrappedData3)).block())); + Flux allActualFlux = structuredMessageEncoder.encode(wrappedData1) + .concatWith(structuredMessageEncoder.encode(wrappedData2)) + .concatWith(structuredMessageEncoder.encode(wrappedData3)); - Assertions.assertArrayEquals(expected, allActualData.toByteArray()); + byte[] actual = FluxUtil.collectBytesInByteBufferStream(allActualFlux).block(); + + assertArrayEquals(expected, actual); } @Test public void emptyBuffer() { StructuredMessageEncoder encoder = new StructuredMessageEncoder(10, 5, StructuredMessageFlags.NONE); ByteBuffer emptyBuffer = ByteBuffer.allocate(0); - ByteBuffer result = ByteBuffer - .wrap(Objects.requireNonNull(collectBytesInByteBufferStream(encoder.encode(emptyBuffer)).block())); - assertEquals(0, result.remaining()); + byte[] result = FluxUtil.collectBytesInByteBufferStream(encoder.encode(emptyBuffer)).block(); + assertNotNull(result); + assertEquals(0, result.length); } @Test public void contentAlreadyEncoded() { StructuredMessageEncoder encoder = new StructuredMessageEncoder(4, 2, StructuredMessageFlags.NONE); - encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2, 3, 4 })).blockLast(); - StepVerifier.create(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2 }))) - .expectError(IllegalArgumentException.class) - .verify(); + FluxUtil.collectBytesInByteBufferStream(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2, 3, 4 }))).block(); + assertThrows(IllegalArgumentException.class, + () -> FluxUtil.collectBytesInByteBufferStream(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2 }))) + .block()); } @Test public void bufferLengthExceedsContentLength() { StructuredMessageEncoder encoder = new StructuredMessageEncoder(4, 2, StructuredMessageFlags.NONE); - encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2, 3 })).blockLast(); - StepVerifier.create(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2 }))) - .expectError(IllegalArgumentException.class) - .verify(); + FluxUtil.collectBytesInByteBufferStream(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2, 3 }))).block(); + assertThrows(IllegalArgumentException.class, + () -> FluxUtil.collectBytesInByteBufferStream(encoder.encode(ByteBuffer.wrap(new byte[] { 1, 2 }))) + .block()); } @Test @@ -248,4 +248,21 @@ public void testNumSegmentsExceedsMaxValue() { assertThrows(IllegalArgumentException.class, () -> new StructuredMessageEncoder(Integer.MAX_VALUE, 1, StructuredMessageFlags.NONE)); } + + @Test + @Disabled("For local testing only") + public void bigEncode() throws IOException { + byte[] data = getRandomData(262144000); + + ByteBuffer unencodedBuffer = ByteBuffer.wrap(data); + + StructuredMessageEncoder structuredMessageEncoder = new StructuredMessageEncoder(262144000, + V1_DEFAULT_SEGMENT_CONTENT_LENGTH, StructuredMessageFlags.STORAGE_CRC64); + + byte[] actual + = FluxUtil.collectBytesInByteBufferStream(structuredMessageEncoder.encode(unencodedBuffer)).block(); + byte[] expected = buildStructuredMessage(unencodedBuffer, V1_DEFAULT_SEGMENT_CONTENT_LENGTH, + StructuredMessageFlags.STORAGE_CRC64).array(); + assertArrayEquals(expected, actual); + } }