Fix broken client body streaming

zio · Sep 30, 2024 · ff4e642 · ff4e642
1 parent dc29b2e
commit ff4e642
Showing 1 changed file with 8 additions and 3 deletions.
diff --git a/zio-http/jvm/src/main/scala/zio/http/netty/AsyncBodyReader.scala b/zio-http/jvm/src/main/scala/zio/http/netty/AsyncBodyReader.scala
@@ -95,12 +95,12 @@ private[netty] abstract class AsyncBodyReader extends SimpleChannelInboundHandle
         onLastMessage()
       }
 
-      val streaming =
+      val readMore =
         state match {
           case State.Buffering                                            =>
             // `connect` method hasn't been called yet, add all incoming content to the buffer
             buffer0.addAll(content)
-            false
+            true
           case State.Direct(callback) if isLast && buffer0.knownSize == 0 =>
             // Buffer is empty, we can just use the array directly
             callback(Chunk.fromArray(content), isLast = true)
@@ -109,14 +109,19 @@ private[netty] abstract class AsyncBodyReader extends SimpleChannelInboundHandle
             // We're aggregating the full response, only call the callback on the last message
             buffer0.addAll(content)
             if (isLast) callback(result(buffer0), isLast = true)
+            !isLast
+          case State.Direct(callback: UnsafeAsync.Streaming)              =>
+            // We're streaming, emit chunks as they come
+            callback(Chunk.fromArray(content), isLast)
+            // ctx.read will be called when the chunk is consumed
             false
           case State.Direct(callback)                                     =>
             // We're streaming, emit chunks as they come
             callback(Chunk.fromArray(content), isLast)
             true
         }
 
-      if (!isLast && !streaming) ctx.read(): Unit
+      if (readMore) ctx.read(): Unit
     }
   }