Merge pull request #353 from normanmaurer/master

Upgrade to Netty 4.0.0.CR7-SNAPSHOT

netty/pom.xml

@@ -12,13 +12,18 @@
     <dependency>
     	<groupId>io.netty</groupId>
     	<artifactId>netty-codec-http</artifactId>
-    	<version>4.0.0.CR1</version>
+    	<version>4.0.0.CR7-SNAPSHOT</version>
     </dependency>
 	 <dependency>
 		<groupId>com.fasterxml.jackson.core</groupId>
 		<artifactId>jackson-databind</artifactId>
 		<version>2.1.1</version>
 	</dependency>
+    <dependency>
+        <groupId>org.javassist</groupId>
+        <artifactId>javassist</artifactId>
+        <version>3.18.0-GA</version>
+    </dependency>
   </dependencies>
   
   <build>

netty/setup.py

@@ -6,7 +6,7 @@ import os
 def start(args):
   try:
     subprocess.check_call("mvn clean compile assembly:single", shell=True, cwd="netty")
-    subprocess.Popen("java -jar netty-example-0.1-jar-with-dependencies.jar".rsplit(" "), cwd="netty/target")
+    subprocess.Popen("java -Dio.netty.noResourceLeakDetection=true -jar netty-example-0.1-jar-with-dependencies.jar".rsplit(" "), cwd="netty/target")
     return 0
   except subprocess.CalledProcessError:
     return 1

netty/src/main/java/hello/HelloServerHandler.java

@@ -1,24 +1,19 @@
 package hello;
 
-import io.netty.buffer.MessageBuf;
+import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundMessageHandlerAdapter;
-import io.netty.handler.codec.DecoderResult;
+import io.netty.channel.MessageList;
+import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.http.Cookie;
 import io.netty.handler.codec.http.CookieDecoder;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.DefaultHttpResponse;
 import io.netty.handler.codec.http.FullHttpResponse;
-import io.netty.handler.codec.http.HttpContent;
 import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.HttpObject;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
-import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.handler.codec.http.ServerCookieEncoder;
-import io.netty.util.CharsetUtil;
 
 import java.util.Map;
 import java.util.HashMap;
@@ -27,90 +22,66 @@ import java.util.Set;
 import java.io.*;
 
 import com.fasterxml.jackson.databind.*;
+import io.netty.util.CharsetUtil;
 
 import static io.netty.handler.codec.http.HttpHeaders.Names.*;
 import static io.netty.handler.codec.http.HttpHeaders.*;
 import static io.netty.handler.codec.http.HttpResponseStatus.*;
 import static io.netty.handler.codec.http.HttpVersion.*;
 
-public class HelloServerHandler extends ChannelInboundMessageHandlerAdapter<Object> {
+public class HelloServerHandler extends SimpleChannelInboundHandler<Object>{
 
-    private HttpRequest request;
     /** Buffer that stores the response content */
-    private final StringBuilder buf = new StringBuilder();
     private static final ObjectMapper mapper = new ObjectMapper();
 
-    private boolean flush;
+    private MessageList<Object> out;
 
     @Override
-    public boolean beginMessageReceived(ChannelHandlerContext ctx) throws Exception {
-        flush = false;
-        return super.beginMessageReceived(ctx);
+    public void beginMessageReceived(ChannelHandlerContext ctx) throws Exception {
+        out = MessageList.newInstance();
+        super.beginMessageReceived(ctx);
     }
 
     @Override
     public void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
-        MessageBuf<Object> out = ctx.nextOutboundMessageBuffer();
         if (msg instanceof HttpRequest) {
-            HttpRequest request = this.request = (HttpRequest) msg;
+            HttpRequest request = (HttpRequest) msg;
 
             if (is100ContinueExpected(request)) {
                 send100Continue(out);
             }
-
+            ByteBuf buf = ctx.alloc().buffer();
             if("/plaintext".equals(request.getUri())) {
-		buf.setLength(0);
-                buf.append("Hello, World!");
+                buf.writeBytes("Hello, World!".getBytes(CharsetUtil.UTF_8));
             } else {
                 Map<String, String> data = new HashMap<String, String>();
                 data.put("message", "Hello, world");
-                
-                buf.setLength(0);
+
                 try
                 {
-                  buf.append(HelloServerHandler.mapper.writeValueAsString(data));
+                  buf.writeBytes(HelloServerHandler.mapper.writeValueAsBytes(data));
                 }
                 catch (IOException ex)
                 {
                   // do nothing
                 }
             }
-
-            appendDecoderResult(buf, request);
+            writeResponse(ctx, request, buf, out);
         }
-
-        if (msg instanceof HttpContent) {
-            if (msg instanceof LastHttpContent) {
-              LastHttpContent trailer = (LastHttpContent) msg;
-              writeResponse(ctx, out, trailer);
-            }
-        }
-    }
-
-    private static void appendDecoderResult(StringBuilder buf, HttpObject o) {
-        DecoderResult result = o.getDecoderResult();
-        if (result.isSuccess()) {
-            return;
-        }
-
-        buf.append(".. WITH DECODER FAILURE: ");
-        buf.append(result.cause());
-        buf.append("\r\n");
     }
 
-    private void writeResponse(ChannelHandlerContext ctx, MessageBuf<Object> out, HttpObject currentObj) {
+    private void writeResponse(ChannelHandlerContext ctx, HttpRequest request, ByteBuf buf, MessageList<Object> out) {
         // Decide whether to close the connection or not.
         boolean keepAlive = isKeepAlive(request);
         // Build the response object.
         FullHttpResponse response = new DefaultFullHttpResponse(
-                HTTP_1_1, currentObj.getDecoderResult().isSuccess()? OK : BAD_REQUEST,
-                Unpooled.copiedBuffer(buf.toString(), CharsetUtil.UTF_8));
+                HTTP_1_1, OK, buf);
 
         response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
 
         if (keepAlive) {
             // Add 'Content-Length' header only for a keep-alive connection.
-            response.headers().set(CONTENT_LENGTH, response.data().readableBytes());
+            response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
             // Add keep alive header as per:
             // - http://www.w3.org/Protocols/HTTP/1.1/draft-ietf-http-v11-spec-01.html#Connection
             response.headers().set(CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
@@ -137,23 +108,22 @@ public class HelloServerHandler extends ChannelInboundMessageHandlerAdapter<Obje
 
         // Close the non-keep-alive connection after the write operation is done.
         if (!keepAlive) {
-            flush = false;
-            ctx.flush().addListener(ChannelFutureListener.CLOSE);
-        } else {
-            flush = true;
+            this.out = MessageList.newInstance();
+            ctx.write(out).addListener(ChannelFutureListener.CLOSE);
         }
     }
 
-    private void send100Continue(MessageBuf<Object> out) {
-        HttpResponse response = new DefaultHttpResponse(HTTP_1_1, CONTINUE);
+    private void send100Continue(MessageList<Object> out) {
+        HttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, CONTINUE, Unpooled.EMPTY_BUFFER);
         out.add(response);
-        flush = true;
     }
 
     @Override
     public void endMessageReceived(ChannelHandlerContext ctx) throws Exception {
-        if (flush) {
-            ctx.flush();
+        if (out != null) {
+            MessageList<Object> msgs = out;
+            this.out = null;
+            ctx.write(msgs);
         }
     }