Netty 4 - Intro → Changes → HTTP → Lessons learned

$ echo "Netty is shit...." | sed -e 's/is /is the /' | cat (1)
                    Netty is the shit....

public class MyChannelInitializer extends ChannelInitializer<Channel> {
  @Override
  public void initChannel(Channel ch) {
    ChannelPipeline p = ch.pipeline();
    p.addLast(new SslHandler(...)); (1)
    p.addLast(new HttpServerCodec(...)); (2)
    p.addLast(new YourRequestHandler()); (3)
  }
}

@Sharable
public class EchoHandler extends ChannelInboundHandlerAdapter {
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) { (1)
    ctx.writeAndFlush(msg);
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { (2)
    cause.printStacktrace();
    ctx.close();
  }
}

public class StringDecoder extends MessageToMessageDecoder<ByteBuf> {
  @Override
  protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) {
      out.add(msg.toString(charset));
  }
}

public class StringEncoder extends MessageToMessageEncoder<String> {
  @Override
  protected void encode(ChannelHandlerContext ctx, String msg, List<Object> out) {
    out.add(ByteBufUtil.encodeString(ctx.alloc(), CharBuffer.wrap(msg), charset));
  }
}

public enum CustomEvents {
  MyCustomEvent
}

public class CustomEventHandler extends ChannelInboundHandlerAdapter {
  @Override
  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
    if (evt == MyCustomEvent) { // do something}
  }
}

ChannelPipeline pipeline = channel.pipeline();
pipeline.fireUserEventTriggered(MyCustomEvent);

EventLoopGroup group = new NioEventLoopGroup();
Bootstrap cb = new Bootstrap();
cb.group(group);

ServerBootstrap sb = new ServerBootstrap();
sb.group(group);

public class WriteTimeOutHandler extends ChannelOutboundHandlerAdapter {
  @Override
  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
    ctx.write(msg, promise);

    if (!promise.isDone()) {
      ctx.executor().schedule(new WriteTimeoutTask(promise), 30, TimeUnit.SECONDS); (1)
    }
  }
}

Channel ch = ...;
ChannelPipeline p = ch.pipeline();
EventExecutor e1 = new DefaultEventExecutor(16);

p.addLast(new MyProtocolCodec()); (1)
p.addLast(e1, new MyDatabaseAccessingHandler()); (2)

Bootstrap bootstrap = new Bootstrap();
bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

public interface ReferenceCounted {
  int refCnt();
  ReferenceCounted retain();
  ReferenceCounted retain(int increment);
  boolean release();
  boolean release(int decrement);
}

public class MyChannelInboundHandler extends ChannelInboundHandlerAdapter {
  public void channelRead(ChannelHandlerContext ctx, Object msg) {
    try {
      ...
    } finally {
      ReferenceCountUtil.release(msg);
    }
  }
}

LEAK: ByteBuf.release() was not called before it's garbage-collected....

LEAK: ByteBuf.release() was not called before it's garbage-collected.
Recent access records: 1
#1:
    io.netty.buffer.AdvancedLeakAwareByteBuf.toString(AdvancedLeakAwareByteBuf.java:697)
    ...
Created at:
    ...
    io.netty.handler.codec.xml.XmlFrameDecoderTest.testDecodeWithXml(XmlFrameDecoderTest.java:147)

Bootstrap bootstrap = new Bootstrap().group(new NioEventLoopGroup());
bootstrap.channel(NioSocketChannel.class);

Bootstrap bootstrap = new Bootstrap().group(new EpollEventLoopGroup());
bootstrap.channel(EpollSocketChannel.class);

ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast(new HttpObjectAggregator(10 * 1024 * 1024)); (1)
pipeline.addLast(new SimpleChannelInboundHandler<FullHttpRequest>() {
  @Override
  public void channelRead(ChannelHandlerContext ctx, FullHttpRequest req) { (2)
    // handle me
  }
});

private static final AsciiString X_HEADER_NAME = new AsciiString("X-Header"); (1)
private static final AsciiString X_VALUE = new AsciiString("Value");

pipeline.addLast(new SimpleChannelInboundHandler<FullHttpRequest>() {
  @Override
  public void channelRead(ChannelHandlerContext ctx, FullHttpRequest req) {
    FullHttpResponse response = new FullHttpResponse(HTTP_1_1, OK);
    response.headers().set(X_HEADER_NAME, X_VALUE); (2)
    ...
    ctx.writeAndFlush(response);
  }
});

RandomAccessFile raf = new RandomAccessFile(file, "r");
HttpResponse response = new FullHttpResponse(HTTP_1_1, OK);
HttpHeaderUtil.setContentLength(response, raf.length());
channel.write(response); (1)
channel.write(new DefaultFileRegion(raf.getChannel(), 0, fileLength)); (2)
channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);  (3)

File f = new File(file, "r");
HttpResponse response = new FullHttpResponse(HTTP_1_1, OK);
HttpHeaders.setContentLength(response, f.length());
channel.write(response); (1)
channel.write(new HttpChunkedInput(new ChunkedNioFile(file))); (2)
channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);  (3)

ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast(new HttpRequestDecoder(4096, 8192, 8192));

HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpStatus.OK);

ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast(new HttpRequestDecoder(4096, 8192, 8192, false));

HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpStatus.OK, false);

public class HttpPipeliningHandler extends SimpleChannelInboundHandler<HttpRequest> {
  @Override
  public void channelRead(ChannelHandlerContext ctx, HttpRequest req) {
    ChannelFuture future = ctx.write(createResponse(req)); (1)
    if (!HttpHeaders.isKeepAlive(req)) {
      future.addListener(ChannelFutureListener.CLOSE); (2)
    }
  }
  @Override
  public void channelReadComplete(ChannelHandlerContext ctx) {
    ctx.flush(); (3)
  }
}

public class HttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
  @Override
  public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest) { (1)
    final Channel inboundChannel = ctx.channel();
    Bootstrap b = new Bootstrap();
    b.group(new NioEventLooopGroup()); (2)
    ...
    ChannelFuture f = b.connect(remoteHost, remotePort);
    ...
  }
}

public class HttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
  @Override
  public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest) { (1)
    final Channel inboundChannel = ctx.channel();
    Bootstrap b = new Bootstrap();
    b.group(inboundChannel.eventLoop()); (2)
    ...
    ChannelFuture f = b.connect(remoteHost, remotePort);
    ...
  }
}

channel.config().setAutoRead(false); (1)
channel.read(); (2)
channel.config().setAutoRead(true); (3)

public class HttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
  @Override public void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest) {
    ctx.channel().setAutoRead(false); (1)
    Bootstrap b = createBootstrap();
    b.connect(remoteHost, remotPort).addListener(new ChannelFutureListener() { (2)
      public void operationComplete(ChannelFuture future) {
        if (future.isSuccess()) ctx.channel().setAutoRead(true); (3)
      }
    });
  }
}

Channel channel = ...;
FileChannel fc = ...;
channel.writeAndFlush(new DefaultFileRegion(fc, 0, fileLength));

pipeline.addLast(new HttpContentCompressor()); (1)

pipeline.addLast(new WebSocketServerProtocolHandler("/ws")); (1)
pipeline.addLast(new SimpleChannelInboundHandler<TextWebSocketFrame>() { (2)
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.writeAndFlush(msg);
  }
});

public class SpdyServerInitializer extends ChannelInitializer<SocketChannel> {
    private final SslContext sslCtx =  ....
    @Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline p = ch.pipeline();
        p.addLast(sslCtx.newHandler(ch.alloc()));
        p.addLast(new SpdyOrHttpChooser() { (1)
          @Override
          protected ChannelInboundHandler createHttpRequestHandlerForHttp() { ... } (2)
          @Override
          protected ChannelInboundHandler createHttpRequestHandlerForSpdy() { ... } (3)
        });
    }
}

static void freeDirectBuffer(ByteBuffer buffer) {
  if (CLEANER_FIELD_OFFSET == -1 || !buffer.isDirect()) {
    return;
  }
  try {
    Cleaner cleaner = (Cleaner) getObject(buffer, CLEANER_FIELD_OFFSET);
    if (cleaner != null) cleaner.clean();

  } catch (Throwable t) { // Nothing we can do here. }
}

int index = -1;
for (int i = buf.readerIndex(); index == -1 && i <  buf.writerIndex(); i++) {
  if (buf.getByte(i) == '\n') {
    index = i;
  }
}

int index = buf.forEachByte(new ByteBufProcessor() {
  @Override
  public boolean process(byte value) {
    return value != '\n';
  }
});