Fix out-of-order write in HttpStreamsHandler

http-netty/src/main/java/io/micronaut/http/netty/stream/HttpStreamsHandler.java

@@ -21,6 +21,7 @@
 import io.micronaut.http.netty.reactive.HandlerPublisher;
 import io.micronaut.http.netty.reactive.HandlerSubscriber;
 import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -391,20 +392,39 @@ protected void unbufferedWrite(final ChannelHandlerContext ctx, final Out messag
             StreamedHttpMessage streamed = (StreamedHttpMessage) message;
             HandlerSubscriber<HttpContent> subscriber = new HandlerSubscriber<HttpContent>(ctx.executor()) {
                 AtomicBoolean messageWritten = new AtomicBoolean();
+                ChannelFuture delayWrites;
 
                 @Override
                 public void onNext(HttpContent httpContent) {
+                    if (delayWrites != null && !delayWrites.isDone()) {
+                        delayWrites.addListener(future -> onNext(httpContent));
+                        return;
+                    }
+
                     if (messageWritten.compareAndSet(false, true)) {
                         ChannelPromise messageWritePromise = ctx.newPromise();
                         //if oncomplete gets called before the message is written the promise
                         //set to lastWriteFuture shouldn't complete until the first content is written
                         lastWriteFuture = messageWritePromise;
-                        ctx.writeAndFlush(message).addListener(f -> super.onNext(httpContent, messageWritePromise));
+                        delayWrites = ctx.writeAndFlush(message);
+                        delayWrites.addListener(f -> {
+                            delayWrites = null;
+                            super.onNext(httpContent, messageWritePromise);
+                        });
                     } else {
                         super.onNext(httpContent);
                     }
                 }
 
+                @Override
+                public void onError(Throwable error) {
+                    if (delayWrites != null && !delayWrites.isDone()) {
+                        delayWrites.addListener(future -> onError(error));
+                        return;
+                    }
+                    super.onError(error);
+                }
+
                 @Override
                 protected void error(Throwable error) {
                     try {
@@ -424,6 +444,15 @@ protected void error(Throwable error) {
                     }
                 }
 
+                @Override
+                public void onComplete() {
+                    if (delayWrites != null && !delayWrites.isDone()) {
+                        delayWrites.addListener(future -> onComplete());
+                        return;
+                    }
+                    super.onComplete();
+                }
+
                 @Override
                 protected void complete() {
                     if (messageWritten.compareAndSet(false, true)) {

http-netty/src/test/groovy/io/micronaut/http/netty/stream/HttpStreamsClientHandlerSpec.groovy

@@ -0,0 +1,66 @@
+package io.micronaut.http.netty.stream
+
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.Unpooled
+import io.netty.channel.ChannelHandlerContext
+import io.netty.channel.ChannelOutboundHandlerAdapter
+import io.netty.channel.ChannelPromise
+import io.netty.channel.embedded.EmbeddedChannel
+import io.netty.handler.codec.http.DefaultHttpContent
+import io.netty.handler.codec.http.DefaultHttpRequest
+import io.netty.handler.codec.http.HttpContent
+import io.netty.handler.codec.http.HttpMethod
+import io.netty.handler.codec.http.HttpRequest
+import io.netty.handler.codec.http.HttpVersion
+import io.netty.handler.codec.http.LastHttpContent
+import reactor.core.publisher.Flux
+import spock.lang.Issue
+import spock.lang.Specification
+
+import java.nio.charset.StandardCharsets
+
+class HttpStreamsClientHandlerSpec extends Specification {
+    @Issue('https://github.com/micronaut-projects/micronaut-tracing/issues/316')
+    def 'out of order write'() {
+        given:
+        ChannelPromise firstWritePromise = null
+        def channel = new EmbeddedChannel(
+                new ChannelOutboundHandlerAdapter() {
+                    @Override
+                    void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+                        // delay the completion of the write of the HttpRequest
+                        if (firstWritePromise == null) {
+                            firstWritePromise = promise
+                            ctx.write(msg, ctx.voidPromise())
+                            return
+                        }
+                        super.write(ctx, msg, promise)
+                    }
+                },
+                new HttpStreamsClientHandler()
+        )
+        def msg = new DelegateStreamedHttpRequest(
+                new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/"),
+                JsonSubscriber.lift(Flux.just(new DefaultHttpContent(Unpooled.wrappedBuffer("\"foo\"".getBytes(StandardCharsets.UTF_8)))))
+        )
+
+        when:
+        channel.writeOutbound(msg)
+        firstWritePromise.trySuccess()
+        channel.flushOutbound()
+        then:
+        channel.readOutbound() instanceof HttpRequest
+        when:
+        ByteBuf combined = Unpooled.buffer()
+        while (true) {
+            HttpContent h = channel.readOutbound()
+            combined.writeBytes(h.content())
+            if (h instanceof LastHttpContent) {
+                break
+            }
+        }
+        then:
+        combined.toString(StandardCharsets.UTF_8) == "[\"foo\"]"
+        !channel.finish()
+    }
+}