Fixed some initialization issues and created reliable ChannelHandler implementations to help deal with UDP reliability

server/netty/src/com/riiablo/server/netty/Client.java

@@ -5,8 +5,10 @@ import com.google.flatbuffers.FlatBufferBuilder;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.SimpleChannelInboundHandler;
@@ -51,7 +53,21 @@ public class Client extends ApplicationAdapter {
           .handler(new ChannelInitializer<DatagramChannel>() {
             @Override
             protected void initChannel(DatagramChannel ch) {
-              ch.pipeline().addLast(new EchoClientHandler());
+              ReliableInboundHandler in = new ReliableInboundHandler();
+              ReliableOutboundHandler out = new ReliableOutboundHandler();
+              final EchoClientHandler echo = new EchoClientHandler();
+              ch.pipeline()
+                  .addLast(in)
+                  .addLast(echo)
+                  .addLast(out)
+                  .addLast(new ChannelInboundHandlerAdapter() {
+                    @Override
+                    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+                      echo.init(ctx.channel());
+                      ctx.pipeline().remove(this);
+                    }
+                  })
+                  ;
             }
           });
 
@@ -69,22 +85,26 @@ public class Client extends ApplicationAdapter {
       super(false);
     }
 
-    @Override
+    void init(Channel ch) {
-    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+      InetSocketAddress remoteAddress = (InetSocketAddress) ch.remoteAddress();
-      InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
       Gdx.app.log(TAG, "Connecting to " + remoteAddress.getHostString() + ":" + remoteAddress.getPort());
 
       FlatBufferBuilder builder = new FlatBufferBuilder();
-      int headerOffset = Header.createHeader(builder, 0, 0, 0);
+      int headerOffset = Header.createHeader(builder, -1, -1, 0);
       Connection.startConnection(builder);
       int dataOffset = Connection.endConnection(builder);
       int offset = Netty.createNetty(builder, headerOffset, NettyData.Connection, dataOffset);
       Netty.finishSizePrefixedNettyBuffer(builder, offset);
 
-//      sanity(builder.dataBuffer());
+      sanity(builder.dataBuffer());
 
       ByteBuf byteBuf = Unpooled.wrappedBuffer(builder.dataBuffer());
-      ctx.writeAndFlush(byteBuf);
+      ch.writeAndFlush(byteBuf);
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
+      ctx.fireChannelRead(msg);
     }
 
     private void sanity(ByteBuffer buffer) {
@@ -95,21 +115,11 @@ public class Client extends ApplicationAdapter {
       Gdx.app.log(TAG, "  " + String.format("SEQ:%d ACK:%d ACK_BITS:%08x", header.sequence(), header.ack(), header.ackBits()));
     }
 
-    @Override
-    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) {
-//      ctx.writeAndFlush(msg);
-      msg.release();
-    }
-
-    @Override
-    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
-//      System.out.println("Read complete.");
-    }
-
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
       Gdx.app.error(TAG, cause.getMessage(), cause);
       ctx.close();
+      ctx.fireExceptionCaught(cause);
     }
   }
 }

server/netty/src/com/riiablo/server/netty/ReliableInboundHandler.java

@@ -0,0 +1,78 @@
+package com.riiablo.server.netty;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandler;
+
+import com.badlogic.gdx.Gdx;
+
+public class ReliableInboundHandler implements ChannelInboundHandler {
+  private static final String TAG = "ReliableInboundHandler";
+
+  ReliableInboundHandler() {}
+
+  @Override
+  public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "channelRegistered");
+    ctx.fireChannelRegistered();
+  }
+
+  @Override
+  public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "channelUnregistered");
+    ctx.fireChannelUnregistered();
+  }
+
+  @Override
+  public void channelActive(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "channelActive");
+    ctx.fireChannelActive();
+  }
+
+  @Override
+  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "channelInactive");
+    ctx.fireChannelInactive();
+  }
+
+  @Override
+  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+    Gdx.app.debug(TAG, "channelRead");
+    ctx.fireChannelRead(msg);
+  }
+
+  @Override
+  public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "channelReadComplete");
+    ctx.fireChannelReadComplete();
+  }
+
+  @Override
+  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+    Gdx.app.debug(TAG, "userEventTriggered");
+    ctx.fireUserEventTriggered(evt);
+  }
+
+  @Override
+  public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "channelWritabilityChanged");
+    ctx.fireChannelWritabilityChanged();
+  }
+
+  @Override
+  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "handlerAdded");
+
+  }
+
+  @Override
+  public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "handlerRemoved");
+
+  }
+
+  @Override
+  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+    Gdx.app.debug(TAG, "exceptionCaught");
+    ctx.fireExceptionCaught(cause);
+  }
+}

server/netty/src/com/riiablo/server/netty/ReliableOutboundHandler.java

@@ -0,0 +1,80 @@
+package com.riiablo.server.netty;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandler;
+import io.netty.channel.ChannelPromise;
+import java.net.SocketAddress;
+
+import com.badlogic.gdx.Gdx;
+
+public class ReliableOutboundHandler implements ChannelOutboundHandler {
+  private static final String TAG = "ReliableOutboundHandler";
+
+  ReliableOutboundHandler() {}
+
+  @Override
+  public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
+    Gdx.app.debug(TAG, "bind");
+    ctx.bind(localAddress, promise);
+  }
+
+  @Override
+  public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
+    Gdx.app.debug(TAG, "connect");
+    ctx.connect(remoteAddress, localAddress, promise);
+  }
+
+  @Override
+  public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+    Gdx.app.debug(TAG, "disconnect");
+    ctx.disconnect(promise);
+  }
+
+  @Override
+  public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+    Gdx.app.debug(TAG, "close");
+    ctx.close(promise);
+  }
+
+  @Override
+  public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+    Gdx.app.debug(TAG, "deregister");
+    ctx.deregister(promise);
+  }
+
+  @Override
+  public void read(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "read");
+    ctx.read();
+  }
+
+  @Override
+  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+    Gdx.app.debug(TAG, "write");
+    ctx.write(msg, promise);
+  }
+
+  @Override
+  public void flush(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "flush");
+    ctx.flush();
+  }
+
+  @Override
+  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "handlerAdded");
+
+  }
+
+  @Override
+  public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+    Gdx.app.debug(TAG, "handlerRemoved");
+
+  }
+
+  @Override
+  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+    Gdx.app.debug(TAG, "channelRegistered");
+    ctx.fireExceptionCaught(cause);
+  }
+}