wip

Author: ibrandes

sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/StorageContentValidationPolicy.java

@@ -107,13 +107,25 @@ private void applyStructuredMessage(HttpPipelineCallContext context) {
         BufferStagingArea stagingArea
             = new BufferStagingArea(STATIC_MAXIMUM_ENCODED_DATA_LENGTH, STATIC_MAXIMUM_ENCODED_DATA_LENGTH);
 
-        Flux<ByteBuffer> encodedBody = context.getHttpRequest()
-            .getBody()
-            .flatMapSequential(stagingArea::write, 1, 1)
-            .concatWith(Flux.defer(stagingArea::flush))
-            .concatMap(bufferAggregator -> bufferAggregator.asFlux().concatMap(structuredMessageEncoder::encode));
+        // reactive stream backpressure will not correctly propagate without limitRate
+        //        Flux<ByteBuffer> encodedBody = Flux.from(context.getHttpRequest().getBody())
+        //            .limitRate(1)  // force one-at-a-time from source
+        //            .flatMapSequential(stagingArea::write, 1, 1)
+        //            .concatWith(Flux.defer(stagingArea::flush))
+        //            .concatMap(bufferAggregator -> bufferAggregator.asFlux().concatMap(structuredMessageEncoder::encode));
+
+//        Flux<ByteBuffer> encodedBody = context.getHttpRequest()
+//            .getBody()
+//            .concatMap(buffer -> splitLargeBuffer(buffer, STATIC_MAXIMUM_ENCODED_DATA_LENGTH))
+//            .limitRate(1)
+//            .concatMap(structuredMessageEncoder::encode, 1);
+
+//        Flux<ByteBuffer> encodedBody = context.getHttpRequest()
+//            .getBody()
+//            .flatMapSequential(stagingArea::write, 1, 1)
+//            .concatWith(Flux.defer(stagingArea::flush))
+//            .concatMap(bufferAggregator -> bufferAggregator.asFlux().concatMap(structuredMessageEncoder::encode));
 
-        // Set the encoded body
         context.getHttpRequest().setBody(encodedBody);
 
         context.getHttpRequest()
@@ -127,4 +139,29 @@ private void applyStructuredMessage(HttpPipelineCallContext context) {
         //return structuredMessageEncoder.getMessageCRC64();
     }
 
+    private Flux<ByteBuffer> splitLargeBuffer(ByteBuffer buffer, int maxChunkSize) {
+        // For small buffers, copy to ensure no retention of larger upstream buffers
+        if (buffer.remaining() <= maxChunkSize) {
+            byte[] copy = new byte[buffer.remaining()];
+            buffer.get(copy);
+            return Flux.just(ByteBuffer.wrap(copy));
+        }
+
+        return Flux.generate(buffer::duplicate, (buf, sink) -> {
+            if (!buf.hasRemaining()) {
+                sink.complete();
+                return buf;
+            }
+
+            int chunkSize = Math.min(buf.remaining(), maxChunkSize);
+
+            // Copy data instead of slicing to avoid retaining original buffer
+            byte[] chunkData = new byte[chunkSize];
+            buf.get(chunkData);
+
+            sink.next(ByteBuffer.wrap(chunkData));
+            return buf;
+        });
+    }
+
 }