netty: refactor outputstream

- it groups write calls
- #3778

Author: jknack

modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyContext.java

@@ -104,7 +104,7 @@ public class NettyContext implements DefaultContext, ChannelFutureListener {
   private String scheme;
   private int port;
   private boolean filesCreated;
-  private NettyHandler connection;
+  NettyHandler connection;
 
   public NettyContext(
       NettyHandler connection,
@@ -530,7 +530,7 @@ public PrintWriter responseWriter(MediaType type) {
   public Sender responseSender() {
     prepareChunked();
     ctx.write(new DefaultHttpResponse(HTTP_1_1, status, setHeaders));
-    return new NettySender(this, ctx);
+    return new NettySender(this);
   }
 
   @NonNull @Override
@@ -583,7 +583,7 @@ Context send(@NonNull ByteBuf data, CharSequence contentLength) {
       responseStarted = true;
       setHeaders.set(CONTENT_LENGTH, contentLength);
       var response = new DefaultFullHttpResponse(HTTP_1_1, status, data, setHeaders, NO_TRAILING);
-      connection.writeHttpObject(response, promise(this));
+      connection.writeMessage(response, promise(this));
       return this;
     } finally {
       requestComplete();
@@ -595,7 +595,7 @@ public Context send(@NonNull ReadableByteChannel channel) {
     try {
       prepareChunked();
       int bufferSize = contentLength > 0 ? (int) contentLength : this.bufferSize;
-      connection.writeHttpChunk(
+      connection.writeChunks(
           new DefaultHttpResponse(HTTP_1_1, status, setHeaders),
           new ChunkedNioStream(channel, bufferSize),
           EMPTY_LAST_CONTENT,
@@ -627,7 +627,7 @@ public Context send(@NonNull InputStream in) {
       long len = responseLength();
       ByteRange range = ByteRange.parse(req.headers().get(RANGE), len).apply(this);
       prepareChunked();
-      connection.writeHttpChunk(
+      connection.writeChunks(
           new DefaultHttpResponse(HTTP_1_1, status, setHeaders),
           new ChunkedStream(range.apply(in), bufferSize),
           EMPTY_LAST_CONTENT,
@@ -670,7 +670,7 @@ public Context send(@NonNull FileChannel file) {
             new HttpChunkedInput(
                 new ChunkedNioFile(file, range.getStart(), range.getEnd(), bufferSize));
 
-        connection.writeHttpChunk(rsp, chunkedInput, promise(this));
+        connection.writeChunks(rsp, chunkedInput, promise(this));
         //        ctx.channel()
         //            .eventLoop()
         //            .execute(
@@ -681,7 +681,7 @@ public Context send(@NonNull FileChannel file) {
         //                  ctx.writeAndFlush(chunkedInput, promise(this));
         //                });
       } else {
-        connection.writeHttpChunk(
+        connection.writeChunks(
             rsp,
             new DefaultFileRegion(file, range.getStart(), range.getEnd()),
             EMPTY_LAST_CONTENT,
@@ -744,7 +744,7 @@ public Context send(@NonNull StatusCode statusCode) {
       var rsp =
           new DefaultFullHttpResponse(
               HTTP_1_1, status, Unpooled.EMPTY_BUFFER, setHeaders, NO_TRAILING);
-      connection.writeHttpObject(rsp, promise(this));
+      connection.writeMessage(rsp, promise(this));
       return this;
     } finally {
       requestComplete();

modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyHandler.java

@@ -36,7 +36,7 @@ public class NettyHandler extends ChannelInboundHandlerAdapter {
   private NettyContext context;
   private boolean read;
   private boolean flush;
-  private ChannelHandlerContext channelContext;
+  ChannelHandlerContext channelContext;
 
   public NettyHandler(
       NettyDateService serverDate,
@@ -128,7 +128,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
     }
   }
 
-  public void writeHttpObject(Object msg, ChannelPromise promise) {
+  public void writeMessage(Object msg, ChannelPromise promise) {
     if (this.channelContext.executor().inEventLoop()) {
       if (this.read) {
         this.flush = true;
@@ -137,31 +137,40 @@ public void writeHttpObject(Object msg, ChannelPromise promise) {
         this.channelContext.writeAndFlush(msg, promise);
       }
     } else {
-      this.channelContext.executor().execute(() -> writeHttpObject(msg, promise));
+      this.channelContext.executor().execute(() -> writeMessage(msg, promise));
     }
   }
 
-  public void writeHttpChunk(Object header, Object body, Object last, ChannelPromise promise) {
+  public void writeChunks(Object header, Object body, Object last, ChannelPromise promise) {
     if (this.channelContext.executor().inEventLoop()) {
       // Headers
-      channelContext.write(header, channelContext.voidPromise());
+      var voidPromise = channelContext.voidPromise();
+      channelContext.write(header, voidPromise);
       // Body
-      channelContext.write(body, channelContext.voidPromise());
+      channelContext.write(body, voidPromise);
       // Finish
       channelContext.writeAndFlush(last, promise);
     } else {
-      this.channelContext.executor().execute(() -> writeHttpChunk(header, body, last, promise));
+      this.channelContext.executor().execute(() -> writeChunks(header, body, last, promise));
+    }
+  }
+
+  public void write(SneakyThrows.Consumer<ChannelHandlerContext> action) {
+    if (this.channelContext.executor().inEventLoop()) {
+      action.accept(this.channelContext);
+    } else {
+      this.channelContext.executor().execute(() -> write(action));
     }
   }
 
-  public void writeHttpChunk(Object header, Object body, ChannelPromise promise) {
+  public void writeChunks(Object header, Object body, ChannelPromise promise) {
     if (this.channelContext.executor().inEventLoop()) {
       // Headers
       channelContext.write(header, channelContext.voidPromise());
       // Body + Last
       channelContext.writeAndFlush(body, promise);
     } else {
-      this.channelContext.executor().execute(() -> writeHttpChunk(header, body, promise));
+      this.channelContext.executor().execute(() -> writeChunks(header, body, promise));
     }
   }
 

modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyOutputStream.java

@@ -20,7 +20,6 @@
 public class NettyOutputStream extends OutputStream {
   private final ByteBuf buffer;
   private final NettyContext ctx;
-  private final ChannelHandlerContext context;
   private final ChannelFutureListener closeListener;
   private HttpResponse headers;
   private final AtomicBoolean closed = new AtomicBoolean(false);
@@ -29,7 +28,6 @@ public NettyOutputStream(
       NettyContext ctx, ChannelHandlerContext context, int bufferSize, HttpResponse headers) {
     this.ctx = ctx;
     this.buffer = context.alloc().heapBuffer(bufferSize, bufferSize);
-    this.context = context;
     this.headers = headers;
     this.closeListener = ctx;
   }
@@ -41,12 +39,6 @@ public void write(int b) {
 
   @Override
   public void write(byte[] src, int off, int len) {
-    write(src, off, len, null);
-  }
-
-  public void write(byte[] src, int off, int len, ChannelFutureListener callback) {
-    writeHeaders();
-
     int dataLengthLeftToWrite = len;
     int dataToWriteOffset = off;
     int spaceLeftInCurrentChunk;
@@ -55,57 +47,65 @@ public void write(byte[] src, int off, int len, ChannelFutureListener callback)
       buffer.writeBytes(src, dataToWriteOffset, spaceLeftInCurrentChunk);
       dataToWriteOffset = dataToWriteOffset + spaceLeftInCurrentChunk;
       dataLengthLeftToWrite = dataLengthLeftToWrite - spaceLeftInCurrentChunk;
-      flush(callback, null);
+      flush(false);
     }
     if (dataLengthLeftToWrite > 0) {
       buffer.writeBytes(src, dataToWriteOffset, dataLengthLeftToWrite);
     }
   }
 
-  private void writeHeaders() {
-    if (headers != null) {
-      context.write(headers, context.voidPromise());
-      headers = null;
-    }
-  }
-
   @Override
   public void flush() throws IOException {
-    flush(null, null);
+    flush(false);
   }
 
-  private void flush(ChannelFutureListener callback, ChannelFutureListener listener) {
+  private void flush(boolean close) {
     int chunkSize = buffer.readableBytes();
     if (chunkSize > 0) {
-      if (listener != null) {
-        if (callback == null) {
-          context.write(new DefaultHttpContent(buffer.copy()), context.voidPromise());
-        } else {
-          context.write(new DefaultHttpContent(buffer.copy())).addListener(callback);
-        }
-        context.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(listener);
-        buffer.release();
+      ByteBuf chunk;
+      if (close) {
+        // don't copy on close
+        chunk = buffer;
       } else {
-        if (callback == null) {
-          context.write(new DefaultHttpContent(buffer.copy()), context.voidPromise());
-        } else {
-          context.write(new DefaultHttpContent(buffer.copy())).addListener(callback);
-        }
+        // make a copy on flush
+        chunk = buffer.copy();
+      }
+      ctx.connection.write(
+          context -> {
+            writeHeaders(context);
+            context.write(new DefaultHttpContent(chunk), context.voidPromise());
+            if (close) {
+              context.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(closeListener);
+            }
+          });
+      if (!close) {
+        // reset
         buffer.clear();
       }
     } else {
-      ChannelFuture future = context.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
-      if (listener != null) {
-        future.addListener(listener);
-      }
+      ctx.connection.write(
+          context -> {
+            writeHeaders(context);
+            ChannelFuture future = context.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+            if (close) {
+              future.addListener(closeListener);
+            }
+          });
+    }
+  }
+
+  private void writeHeaders(ChannelHandlerContext context) {
+    if (headers != null) {
+      context.write(headers, context.voidPromise());
+      headers = null;
     }
   }
 
   @Override
   public void close() {
     if (closed.compareAndSet(false, true)) {
       try {
-        flush(null, closeListener);
+        flush(true);
       } finally {
         ctx.requestComplete();
       }

modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettySender.java

@@ -21,9 +21,9 @@ public class NettySender implements Sender {
   private final NettyContext ctx;
   private final ChannelHandlerContext context;
 
-  public NettySender(NettyContext ctx, ChannelHandlerContext context) {
+  public NettySender(NettyContext ctx) {
     this.ctx = ctx;
-    this.context = context;
+    this.context = ctx.ctx;
   }
 
   @Override