Unmute channel when flush last http stream chunk (elastic#113222)

Tim-Brooks · Oct 29, 2024 · ab61817 · ab61817
1 parent 003bea2
commit ab61817
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 4 deletions.
diff --git a/...sport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java b/...sport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java
@@ -87,10 +87,6 @@ public void handleNettyContent(HttpContent httpContent) {
                 send();
             }
         }
-        if (hasLast) {
-            channel.config().setAutoRead(true);
-            channel.closeFuture().removeListener(closeListener);
-        }
     }
 
     // adds chunk to current buffer, will allocate composite buffer when need to hold more than 1 chunk
@@ -133,6 +129,10 @@ private void send() {
             tracer.onNext(bytesRef, hasLast);
         }
         handler.onNext(bytesRef, hasLast);
+        if (hasLast) {
+            channel.config().setAutoRead(true);
+            channel.closeFuture().removeListener(closeListener);
+        }
     }
 
     @Override

diff --git a/...-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStreamTests.java b/...-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStreamTests.java
@@ -82,6 +82,15 @@ public void testFlushAllReceivedChunks() {
         assertEquals(chunkSize * totalChunks, totalBytes.get());
     }
 
+    // ensures that channel.setAutoRead(true) only when we flush last chunk
+    public void testSetAutoReadOnLastFlush() {
+        channel.writeInbound(randomLastContent(10));
+        assertFalse("should not auto-read on last content reception", channel.config().isAutoRead());
+        stream.next();
+        channel.runPendingTasks();
+        assertTrue("should set auto-read once last content is flushed", channel.config().isAutoRead());
+    }
+
     // ensures that we read from channel when no current chunks available
     // and pass next chunk downstream without holding
     public void testReadFromChannel() {