Handle keep alive in echo server

Author: rdegnan

reactivesocket-netty/src/examples/java/io/reactivesocket/netty/EchoServer.java

@@ -28,7 +28,7 @@ protected void initChannel(Channel ch) throws Exception {
                         }
                     });
 
-            Channel localhost = b.bind("localhost", 8025).sync().channel();
+            Channel localhost = b.bind("0.0.0.0", 8025).sync().channel();
             localhost.closeFuture().sync();
         } finally {
             bossGroup.shutdownGracefully();

reactivesocket-netty/src/examples/java/io/reactivesocket/netty/EchoServerHandler.java

@@ -54,7 +54,7 @@ private static boolean isHttp(int magic1, int magic2) {
     private void switchToHttp(ChannelHandlerContext ctx) {
         ChannelPipeline p = ctx.pipeline();
         p.addLast(new HttpServerCodec());
-        p.addLast(new HttpObjectAggregator(64 * 1024));
+        p.addLast(new HttpObjectAggregator(256 * 1024));
         p.addLast(httpHandler);
         p.remove(this);
     }

reactivesocket-netty/src/examples/java/io/reactivesocket/netty/HttpServerHandler.java

@@ -1,24 +1,22 @@
 package io.reactivesocket.netty;
 
+import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.FullHttpRequest;
-
-import static io.netty.handler.codec.http.HttpResponseStatus.OK;
-import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import io.netty.handler.codec.http.*;
 
 @ChannelHandler.Sharable
 public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
     @Override
-    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
-        ctx.flush();
-        ctx.close();
-    }
-
-    @Override
-    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
-        ctx.write(new DefaultFullHttpResponse(HTTP_1_1, OK, msg.content().copy()));
+    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
+        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, request.content().retain());
+        HttpUtil.setContentLength(response, response.content().readableBytes());
+        if (HttpUtil.isKeepAlive(request)) {
+            response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
+            ctx.writeAndFlush(response);
+        } else {
+            ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+        }
     }
 }