Ensure onNext will be called even if publishing empty content and onC…

.changes/next-release/bugfix-AWSSDKforJavav2-e70484b.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Sends final checksum chunk and trailer when only onComplete() is called by upstream (empty content)"
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ChecksumCalculatingAsyncRequestBody.java

@@ -16,7 +16,8 @@
 package software.amazon.awssdk.core.internal.async;
 
 import static software.amazon.awssdk.core.HttpChecksumConstant.DEFAULT_ASYNC_CHUNK_SIZE;
-import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.calculateChecksumContentLength;
+import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.LAST_CHUNK_LEN;
+import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.calculateChecksumTrailerLength;
 import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.calculateChunkLength;
 import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.createChecksumTrailer;
 import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.createChunk;
@@ -28,11 +29,13 @@
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
+import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.core.checksums.Algorithm;
 import software.amazon.awssdk.core.checksums.SdkChecksum;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.utils.BinaryUtils;
 import software.amazon.awssdk.utils.Validate;
+import software.amazon.awssdk.utils.async.DelegatingSubscriber;
 import software.amazon.awssdk.utils.builder.SdkBuilder;
 
 /**
@@ -129,13 +132,12 @@ public ChecksumCalculatingAsyncRequestBody.Builder trailerHeader(String trailerH
 
     @Override
     public Optional<Long> contentLength() {
-
         if (wrapped.contentLength().isPresent() && algorithm != null) {
             return Optional.of(calculateChunkLength(wrapped.contentLength().get())
-                    + calculateChecksumContentLength(algorithm, trailerHeader));
-        } else {
-            return wrapped.contentLength();
+                               + LAST_CHUNK_LEN
+                               + calculateChecksumTrailerLength(algorithm, trailerHeader));
         }
+        return wrapped.contentLength();
     }
 
     @Override
@@ -149,12 +151,15 @@ public void subscribe(Subscriber<? super ByteBuffer> s) {
         if (sdkChecksum != null) {
             sdkChecksum.reset();
         }
-
         SynchronousChunkBuffer synchronousChunkBuffer = new SynchronousChunkBuffer(totalBytes);
-        wrapped.flatMapIterable(synchronousChunkBuffer::buffer)
+        alwaysInvokeOnNext(wrapped).flatMapIterable(synchronousChunkBuffer::buffer)
                .subscribe(new ChecksumCalculatingSubscriber(s, sdkChecksum, trailerHeader, totalBytes));
     }
 
+    private SdkPublisher<ByteBuffer> alwaysInvokeOnNext(SdkPublisher<ByteBuffer> source) {
+        return subscriber -> source.subscribe(new OnNextGuaranteedSubscriber(subscriber));
+    }
+
     private static final class ChecksumCalculatingSubscriber implements Subscriber<ByteBuffer> {
 
         private final Subscriber<? super ByteBuffer> wrapped;
@@ -243,4 +248,30 @@ private Iterable<ByteBuffer> buffer(ByteBuffer bytes) {
         }
     }
 
+    public static class OnNextGuaranteedSubscriber extends DelegatingSubscriber<ByteBuffer, ByteBuffer> {
+
+        private volatile boolean onNextInvoked;
+
+        public OnNextGuaranteedSubscriber(Subscriber<? super ByteBuffer> subscriber) {
+            super(subscriber);
+        }
+
+        @Override
+        public void onNext(ByteBuffer t) {
+            if (!onNextInvoked) {
+                onNextInvoked = true;
+            }
+
+            subscriber.onNext(t);
+        }
+
+        @Override
+        public void onComplete() {
+            if (!onNextInvoked) {
+                subscriber.onNext(ByteBuffer.wrap(new byte[0]));
+            }
+            super.onComplete();
+        }
+    }
+
 }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/HttpChecksumStage.java

@@ -20,8 +20,8 @@
 import static software.amazon.awssdk.core.HttpChecksumConstant.DEFAULT_ASYNC_CHUNK_SIZE;
 import static software.amazon.awssdk.core.HttpChecksumConstant.SIGNING_METHOD;
 import static software.amazon.awssdk.core.internal.io.AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE;
-import static software.amazon.awssdk.core.internal.io.AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength;
-import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.calculateChecksumContentLength;
+import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.calculateChecksumTrailerLength;
+import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.calculateStreamContentLength;
 import static software.amazon.awssdk.core.internal.util.HttpChecksumResolver.getResolvedChecksumSpecs;
 import static software.amazon.awssdk.http.Header.CONTENT_LENGTH;
 
@@ -179,7 +179,7 @@ private void addFlexibleChecksumInTrailer(SdkHttpFullRequest.Builder request, Re
             }
         }
 
-        long checksumContentLength = calculateChecksumContentLength(checksumSpecs.algorithm(), checksumSpecs.headerName());
+        long checksumContentLength = calculateChecksumTrailerLength(checksumSpecs.algorithm(), checksumSpecs.headerName());
         long contentLen = checksumContentLength + calculateStreamContentLength(originalContentLength, chunkSize);
 
         request.putHeader(HttpChecksumConstant.HEADER_FOR_TRAILER_REFERENCE, checksumSpecs.headerName())

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/io/AwsUnsignedChunkedEncodingInputStream.java

@@ -18,7 +18,6 @@
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.core.checksums.Algorithm;
 import software.amazon.awssdk.core.checksums.SdkChecksum;
 import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.core.internal.chunked.AwsChunkedEncodingConfig;
@@ -40,48 +39,6 @@ public static Builder builder() {
         return new Builder();
     }
 
-    /**
-     * Calculates the content length for a given Algorithm and header name.
-     *
-     * @param algorithm  Algorithm used.
-     * @param headerName Header name.
-     * @return Content length of the trailer that will be appended at the end.
-     */
-    public static long calculateChecksumContentLength(Algorithm algorithm, String headerName) {
-        return headerName.length()
-               + HEADER_COLON_SEPARATOR.length()
-               + algorithm.base64EncodedLength().longValue()
-               + CRLF.length() + CRLF.length();
-    }
-
-    /**
-     *
-     * @param originalContentLength Original Content length.
-     * @return Calculatec Chunk Length with the chunk encoding format.
-     */
-    private static long calculateChunkLength(long originalContentLength) {
-        return Long.toHexString(originalContentLength).length()
-                + CRLF.length()
-                + originalContentLength
-                + CRLF.length();
-    }
-
-    public static long calculateStreamContentLength(long originalLength, long defaultChunkSize) {
-        if (originalLength < 0 || defaultChunkSize == 0) {
-            throw new IllegalArgumentException(originalLength + ", " + defaultChunkSize + "Args <= 0 not expected");
-        }
-
-        long maxSizeChunks = originalLength / defaultChunkSize;
-        long remainingBytes = originalLength % defaultChunkSize;
-
-        long allChunks = maxSizeChunks * calculateChunkLength(defaultChunkSize);
-        long remainingInChunk = remainingBytes > 0 ? calculateChunkLength(remainingBytes) : 0;
-        // last byte is composed of a "0" and "\r\n"
-        long lastByteSize = 1 + (long) CRLF.length();
-
-        return allChunks +  remainingInChunk + lastByteSize;
-    }
-
     @Override
     protected byte[] createFinalChunk(byte[] finalChunk) {
         StringBuilder chunkHeader = new StringBuilder();

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/util/ChunkContentUtils.java

@@ -28,35 +28,64 @@ public final class ChunkContentUtils {
     public static final String ZERO_BYTE = "0";
     public static final String CRLF = "\r\n";
 
+    public static final String LAST_CHUNK = ZERO_BYTE + CRLF;
+    public static final long LAST_CHUNK_LEN = LAST_CHUNK.length();
+
     private ChunkContentUtils() {
     }
 
     /**
+     * The chunk format is: chunk-size CRLF chunk-data CRLF.
+     *
      * @param originalContentLength Original Content length.
-     * @return Calculates Chunk Length.
+     * @return the length of this chunk
      */
     public static long calculateChunkLength(long originalContentLength) {
+        if (originalContentLength == 0) {
+            return 0;
+        }
         return Long.toHexString(originalContentLength).length()
-                + CRLF.length()
-                + originalContentLength
-                + CRLF.length()
-                + ZERO_BYTE.length() + CRLF.length();
+               + CRLF.length()
+               + originalContentLength
+               + CRLF.length();
+    }
+
+    /**
+     * Calculates the content length for data that is divided into chunks.
+     *
+     * @param originalLength  original content length.
+     * @param chunkSize chunk size
+     * @return Content length of the trailer that will be appended at the end.
+     */
+    public static long calculateStreamContentLength(long originalLength, long chunkSize) {
+        if (originalLength < 0 || chunkSize == 0) {
+            throw new IllegalArgumentException(originalLength + ", " + chunkSize + "Args <= 0 not expected");
+        }
+
+        long maxSizeChunks = originalLength / chunkSize;
+        long remainingBytes = originalLength % chunkSize;
+
+        long allChunks = maxSizeChunks * calculateChunkLength(chunkSize);
+        long remainingInChunk = remainingBytes > 0 ? calculateChunkLength(remainingBytes) : 0;
+        // last byte is composed of a "0" and "\r\n"
+        long lastByteSize = 1 + (long) CRLF.length();
+
+        return allChunks +  remainingInChunk + lastByteSize;
     }
 
     /**
-     * Calculates the content length for a given Algorithm and header name.
+     * Calculates the content length for a given algorithm and header name.
      *
      * @param algorithm  Algorithm used.
      * @param headerName Header name.
      * @return Content length of the trailer that will be appended at the end.
      */
-    public static long calculateChecksumContentLength(Algorithm algorithm, String headerName) {
-        int checksumLength = algorithm.base64EncodedLength();
-
-        return (headerName.length()
-                + HEADER_COLON_SEPARATOR.length()
-                + checksumLength
-                + CRLF.length() + CRLF.length());
+    public static long calculateChecksumTrailerLength(Algorithm algorithm, String headerName) {
+        return headerName.length()
+               + HEADER_COLON_SEPARATOR.length()
+               + algorithm.base64EncodedLength().longValue()
+               + CRLF.length()
+               + CRLF.length();
     }
 
     /**
@@ -86,17 +115,13 @@ public static ByteBuffer createChunk(ByteBuffer chunkData, boolean isLastByte) {
         chunkHeader.append(CRLF);
         try {
             byte[] header = chunkHeader.toString().getBytes(StandardCharsets.UTF_8);
-            // Last byte does not need additional \r\n trailer
             byte[] trailer = !isLastByte ? CRLF.getBytes(StandardCharsets.UTF_8)
                     : "".getBytes(StandardCharsets.UTF_8);
             ByteBuffer chunkFormattedBuffer = ByteBuffer.allocate(header.length + chunkLength + trailer.length);
-            chunkFormattedBuffer.put(header)
-                    .put(chunkData)
-                    .put(trailer);
+            chunkFormattedBuffer.put(header).put(chunkData).put(trailer);
             chunkFormattedBuffer.flip();
             return chunkFormattedBuffer;
         } catch (Exception e) {
-            // This is to warp BufferOverflowException,ReadOnlyBufferException to SdkClientException.
             throw SdkClientException.builder()
                     .message("Unable to create chunked data. " + e.getMessage())
                     .cause(e)

core/sdk-core/src/test/java/software/amazon/awssdk/core/checksum/AwsChunkedEncodingInputStreamTest.java

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.core.checksum;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static software.amazon.awssdk.core.internal.util.ChunkContentUtils.calculateChecksumTrailerLength;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -25,6 +26,7 @@
 import software.amazon.awssdk.core.checksums.SdkChecksum;
 import software.amazon.awssdk.core.internal.io.AwsChunkedEncodingInputStream;
 import software.amazon.awssdk.core.internal.io.AwsUnsignedChunkedEncodingInputStream;
+import software.amazon.awssdk.core.internal.util.ChunkContentUtils;
 
 public class AwsChunkedEncodingInputStreamTest {
 
@@ -55,10 +57,9 @@ public void readAwsUnsignedChunkedEncodingInputStream() throws IOException {
     public void lengthsOfCalculateByChecksumCalculatingInputStream(){
 
         String initialString = "Hello world";
-        long calculateChunkLength = AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(initialString.length(),
-                                                                                                       AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE);
-        long checksumContentLength = AwsUnsignedChunkedEncodingInputStream.calculateChecksumContentLength(
-                SHA256_ALGORITHM, SHA256_HEADER_NAME);
+        long calculateChunkLength = ChunkContentUtils.calculateStreamContentLength(initialString.length(),
+                                                                                   AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE);
+        long checksumContentLength = calculateChecksumTrailerLength(SHA256_ALGORITHM, SHA256_HEADER_NAME);
         assertThat(calculateChunkLength).isEqualTo(19);
         assertThat(checksumContentLength).isEqualTo(71);
     }