Changing LAST_HTTP_CONTENT_RECEIVED_KEY key name to STREAMING_COMPLETE_KEY and setting it irrespective of ignoreBodyRead flag

Author: joviegas

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ChannelAttributeKey.java

@@ -74,10 +74,11 @@ public final class ChannelAttributeKey {
         "aws.http.nio.netty.async.channelDiagnostics");
 
     /**
-     * {@link AttributeKey} to keep track of whether we have received the {@link LastHttpContent}.
+     * {@link AttributeKey} to keep track of whether the streaming is completed and this is set to true when we receive the *
+     * {@link LastHttpContent}.
      */
-    public static final AttributeKey<Boolean> LAST_HTTP_CONTENT_RECEIVED_KEY = NettyUtils.getOrCreateAttributeKey(
-        "aws.http.nio.netty.async.lastHttpContentReceived");
+    public static final AttributeKey<Boolean> STREAMING_COMPLETE_KEY = NettyUtils.getOrCreateAttributeKey(
+        "aws.http.nio.netty.async.streamingCompleted");
 
     /**
      * {@link AttributeKey} to keep track of whether we should close the connection after this request

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/NettyRequestExecutor.java

@@ -21,7 +21,7 @@
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.EXECUTION_ID_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.IN_USE;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.KEEP_ALIVE;
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.LAST_HTTP_CONTENT_RECEIVED_KEY;
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.STREAMING_COMPLETE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.REQUEST_CONTEXT_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_COMPLETE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_CONTENT_LENGTH;
@@ -195,7 +195,7 @@ private void configureChannel() {
         channel.attr(EXECUTE_FUTURE_KEY).set(executeFuture);
         channel.attr(REQUEST_CONTEXT_KEY).set(context);
         channel.attr(RESPONSE_COMPLETE_KEY).set(false);
-        channel.attr(LAST_HTTP_CONTENT_RECEIVED_KEY).set(false);
+        channel.attr(STREAMING_COMPLETE_KEY).set(false);
         channel.attr(RESPONSE_CONTENT_LENGTH).set(null);
         channel.attr(RESPONSE_DATA_READ).set(null);
         channel.attr(CHANNEL_DIAGNOSTICS).get().incrementRequestCount();

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ResponseHandler.java

@@ -19,7 +19,7 @@
 import static java.util.stream.Collectors.mapping;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.EXECUTE_FUTURE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.KEEP_ALIVE;
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.LAST_HTTP_CONTENT_RECEIVED_KEY;
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.STREAMING_COMPLETE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.REQUEST_CONTEXT_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_COMPLETE_KEY;
 import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.RESPONSE_CONTENT_LENGTH;
@@ -463,10 +463,10 @@ public void cancel() {
     private void notifyIfResponseNotCompleted(ChannelHandlerContext handlerCtx) {
         RequestContext requestCtx = handlerCtx.channel().attr(REQUEST_CONTEXT_KEY).get();
         Boolean responseCompleted = handlerCtx.channel().attr(RESPONSE_COMPLETE_KEY).get();
-        Boolean lastHttpContentReceived = handlerCtx.channel().attr(LAST_HTTP_CONTENT_RECEIVED_KEY).get();
+        Boolean isStreamingComplete = handlerCtx.channel().attr(STREAMING_COMPLETE_KEY).get();
         handlerCtx.channel().attr(KEEP_ALIVE).set(false);
 
-        if (!Boolean.TRUE.equals(responseCompleted) && !Boolean.TRUE.equals(lastHttpContentReceived)) {
+        if (!Boolean.TRUE.equals(responseCompleted) && !Boolean.TRUE.equals(isStreamingComplete)) {
             IOException err = new IOException(NettyUtils.closedChannelMessage(handlerCtx.channel()));
             runAndLogError(handlerCtx.channel(), () -> "Fail to execute SdkAsyncHttpResponseHandler#onError",
                            () -> requestCtx.handler().onError(err));

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/nrs/HttpStreamsHandler.java

@@ -15,7 +15,7 @@
 
 package software.amazon.awssdk.http.nio.netty.internal.nrs;
 
-import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.LAST_HTTP_CONTENT_RECEIVED_KEY;
+import static software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey.STREAMING_COMPLETE_KEY;
 
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelFuture;
@@ -213,9 +213,12 @@ private void handleCancelled(ChannelHandlerContext ctx, InT msg) {
     }
 
     private void handleReadHttpContent(ChannelHandlerContext ctx, HttpContent content) {
+        boolean lastHttpContent = content instanceof LastHttpContent;
+        if(lastHttpContent){
+            ctx.channel().attr(STREAMING_COMPLETE_KEY).set(true);
+        }
         if (!ignoreBodyRead) {
-            if (content instanceof LastHttpContent) {
-                ctx.channel().attr(LAST_HTTP_CONTENT_RECEIVED_KEY).set(true);
+            if (lastHttpContent) {
                 logger.debug(ctx.channel(), () -> "Received LastHttpContent " + ctx.channel());
                 if (content.content().readableBytes() > 0 ||
                         !((LastHttpContent) content).trailingHeaders().isEmpty()) {
@@ -235,7 +238,7 @@ private void handleReadHttpContent(ChannelHandlerContext ctx, HttpContent conten
 
         } else {
             ReferenceCountUtil.release(content);
-            if (content instanceof LastHttpContent) {
+            if (lastHttpContent) {
                 ignoreBodyRead = false;
                 if (currentlyStreamedMessage != null) {
                     removeHandlerIfActive(ctx, ctx.name() + "-body-publisher");