JERSEY-3163: NettyResponseWriter must not close the connection/channel on every request

test: start netty-helloworld and execute

$ curl -v localhost:8080/helloworld localhost:8080/helloworld

Change-Id: I40039645369c06061f7cb5df86034f1cf6aa904d

Author: pavelbucek

connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/internal/JerseyChunkedInput.java

@@ -192,16 +192,7 @@ public ByteBuffer get() {
 
     @Override
     public void write(final byte[] b) throws IOException {
-
-        final byte[] bytes = new byte[b.length];
-        System.arraycopy(b, 0, bytes, 0, b.length);
-
-        write(new Provider<ByteBuffer>() {
-            @Override
-            public ByteBuffer get() {
-                return ByteBuffer.wrap(bytes);
-            }
-        });
+        write(b, 0, b.length);
     }
 
     @Override

connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/internal/NettyInputStream.java

@@ -64,6 +64,11 @@ public class NettyInputStream extends InputStream {
         public int read() throws IOException {
             return 0;
         }
+
+        @Override
+        public String toString() {
+            return "END_OF_INPUT " + super.toString();
+        }
     };
 
     /**
@@ -74,6 +79,11 @@ public int read() throws IOException {
         public int read() throws IOException {
             return 0;
         }
+
+        @Override
+        public String toString() {
+            return "END_OF_INPUT_ERROR " + super.toString();
+        }
     };
 
     private final LinkedBlockingDeque<InputStream> isList;

containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyServerHandler.java

@@ -103,6 +103,7 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) {
                 ctx.write(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE));
             }
 
+            isList.clear(); // clearing the content - possible leftover from previous request processing.
             final ContainerRequest requestContext = createContainerRequest(ctx, req);
 
             requestContext.setWriter(new NettyResponseWriter(ctx, req, container));

containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/NettyHttp2ResponseWriter.java

@@ -54,6 +54,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpResponseStatus;
@@ -134,14 +135,12 @@ public void flush() throws IOException {
 
                 @Override
                 public void close() throws IOException {
-                    ctx.writeAndFlush(new DefaultHttp2DataFrame(true));
-                    ctx.close();
+                    ctx.write(new DefaultHttp2DataFrame(true)).addListener(NettyResponseWriter.FLUSH_FUTURE);
                 }
             };
 
         } else {
             ctx.writeAndFlush(new DefaultHttp2DataFrame(true));
-            ctx.close();
             return null;
         }
     }

containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/NettyResponseWriter.java

@@ -45,8 +45,15 @@
 import java.util.Map;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
-import io.netty.buffer.Unpooled;
+import org.glassfish.jersey.netty.connector.internal.JerseyChunkedInput;
+import org.glassfish.jersey.server.ContainerException;
+import org.glassfish.jersey.server.ContainerResponse;
+import org.glassfish.jersey.server.spi.ContainerResponseWriter;
+
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
@@ -58,10 +65,7 @@
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpUtil;
-import org.glassfish.jersey.netty.connector.internal.JerseyChunkedInput;
-import org.glassfish.jersey.server.ContainerException;
-import org.glassfish.jersey.server.ContainerResponse;
-import org.glassfish.jersey.server.spi.ContainerResponseWriter;
+import io.netty.handler.codec.http.LastHttpContent;
 
 /**
  * Netty implementation of {@link ContainerResponseWriter}.
@@ -70,23 +74,41 @@
  */
 class NettyResponseWriter implements ContainerResponseWriter {
 
+    private static final Logger LOGGER = Logger.getLogger(NettyResponseWriter.class.getName());
+
+    static final ChannelFutureListener FLUSH_FUTURE = new ChannelFutureListener() {
+        @Override
+        public void operationComplete(ChannelFuture future) throws Exception {
+            future.channel().flush();
+        }
+    };
+
     private final ChannelHandlerContext ctx;
     private final HttpRequest req;
     private final NettyHttpContainer container;
 
     private volatile ScheduledFuture<?> suspendTimeoutFuture;
     private volatile Runnable suspendTimeoutHandler;
 
+    private boolean responseWritten = false;
+
     NettyResponseWriter(ChannelHandlerContext ctx, HttpRequest req, NettyHttpContainer container) {
         this.ctx = ctx;
         this.req = req;
         this.container = container;
     }
 
     @Override
-    public OutputStream writeResponseStatusAndHeaders(long contentLength, ContainerResponse responseContext)
+    public synchronized OutputStream writeResponseStatusAndHeaders(long contentLength, ContainerResponse responseContext)
             throws ContainerException {
 
+        if (responseWritten) {
+            LOGGER.log(Level.FINE, "Response already written.");
+            return null;
+        }
+
+        responseWritten = true;
+
         String reasonPhrase = responseContext.getStatusInfo().getReasonPhrase();
         int statusCode = responseContext.getStatus();
 
@@ -122,14 +144,14 @@ public OutputStream writeResponseStatusAndHeaders(long contentLength, ContainerR
             JerseyChunkedInput jerseyChunkedInput = new JerseyChunkedInput(ctx.channel());
 
             if (HttpUtil.isTransferEncodingChunked(response)) {
-                ctx.write(new HttpChunkedInput(jerseyChunkedInput)).addListener(ChannelFutureListener.CLOSE);
+                ctx.write(new HttpChunkedInput(jerseyChunkedInput)).addListener(FLUSH_FUTURE);
             } else {
-                ctx.write(jerseyChunkedInput).addListener(ChannelFutureListener.CLOSE);
+                ctx.write(new HttpChunkedInput(jerseyChunkedInput)).addListener(FLUSH_FUTURE);
             }
             return jerseyChunkedInput;
 
         } else {
-            ctx.write(Unpooled.buffer(0)).addListener(ChannelFutureListener.CLOSE);
+            ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
             return null;
         }
     }
@@ -178,8 +200,7 @@ public void commit() {
 
     @Override
     public void failure(Throwable error) {
-        ctx.writeAndFlush(
-                new DefaultFullHttpResponse(req.protocolVersion(), HttpResponseStatus.INTERNAL_SERVER_ERROR))
+        ctx.writeAndFlush(new DefaultFullHttpResponse(req.protocolVersion(), HttpResponseStatus.INTERNAL_SERVER_ERROR))
            .addListener(ChannelFutureListener.CLOSE);
     }
 

pom.xml

@@ -1943,7 +1943,7 @@
         <mimepull.version>1.9.6</mimepull.version>
         <moxy.version>2.6.0</moxy.version>
         <mustache.version>0.8.17</mustache.version>
-        <netty.version>4.1.4.Final</netty.version>
+        <netty.version>4.1.5.Final</netty.version>
         <osgi.version>4.2.0</osgi.version>
         <pax.exam.version>4.9.1</pax.exam.version>
         <pax.web.version>0.7.4</pax.web.version><!-- TODO: UPGRADE! -->