title: Creating a one-to-many TCP proxy using Netty
body: <p>I'm trying to create a TCP proxy that forwards request to many other TCP endpoints using Netty/Java.</p>
  
  <p>For example:</p>
  
  <pre><code>                     /--&gt; SERVER A 
  Client A --&gt; PROXY --
                       \--&gt; SERVER B 
  </code></pre>
  
  <p>If <code>Client A</code> sends a TCP command through the proxy, the proxy opens two TCP connections to <code>Server A</code> and <code>Server B</code>, and concurrently proxies the request sent by <code>Client A</code> to both of them. </p>
  
  <p>If <code>Client A</code> subsequently sends another command, the proxy theoretically has previously cached the two connections in a pool, so without opening two new connections again, proxies the request to the two servers.</p>
  
  <p>Regarding the response handling, I would like to have two options:</p>
  
  <ul>
  <li>Show two responses one after the other to <code>Client A</code>.</li>
  <li>Or completely ignore the response.</li>
  </ul>
  
  <p>If a connection is lost or closed, the proxy should be able to automatically recreate it and add it back to the connection pool.</p>
  
  <p>I've been looking at the Netty examples, and tried to use <a href="http://netty.io/4.0/api/io/netty/channel/group/ChannelGroup.html" rel="nofollow"><code>ChannelGroup</code></a> to handle the connection pool, but without success. Also, in my code below, after sending the first request the proxy stops working. Any tips?</p>
  
  <pre><code>import io.netty.bootstrap.Bootstrap;
  import io.netty.buffer.Unpooled;
  import io.netty.channel.Channel;
  import io.netty.channel.ChannelFuture;
  import io.netty.channel.ChannelFutureListener;
  import io.netty.channel.ChannelHandlerContext;
  import io.netty.channel.ChannelInboundHandlerAdapter;
  import io.netty.channel.ChannelOption;
  import io.netty.channel.group.ChannelGroup;
  import io.netty.channel.group.DefaultChannelGroup;
  import io.netty.util.concurrent.GlobalEventExecutor;
  
  import java.util.LinkedList;
  import java.util.List;
  
  public class TcpProxyHandler extends ChannelInboundHandlerAdapter {
  
      private static List&lt;String&gt; hosts = new LinkedList&lt;&gt;();
      private static List&lt;String&gt; connected = new LinkedList&lt;&gt;();
  
      static {
          hosts.add("127.0.0.1:10000");
          hosts.add("127.0.0.1:20000");
      }
  
      static final ChannelGroup channels = new DefaultChannelGroup(
              GlobalEventExecutor.INSTANCE);
  
      @Override
      public void channelActive(ChannelHandlerContext ctx) throws Exception {
          final Channel inboundChannel = ctx.channel();
  
          for (String host : hosts) {
              if (!connected.contains(host)) {
                  String address = host.split(":")[0];
                  int port = Integer.parseInt(host.split(":")[1]);
                  Channel outboundChannel = ConnectionPool.getConnection(address,
                          port);
                  if (outboundChannel == null) {
                      Bootstrap b = new Bootstrap();
                      b.group(inboundChannel.eventLoop())
                              .channel(ctx.channel().getClass())
                              .handler(new TcpProxyBackendHandler(inboundChannel))
                              .option(ChannelOption.AUTO_READ, false);
                      ChannelFuture f = b.connect(address, port);
                      outboundChannel = f.channel();
                      f.addListener(new ChannelFutureListener() {
                          @Override
                          public void operationComplete(ChannelFuture future)
                                  throws Exception {
                              if (future.isSuccess()) {
                                  // connection complete start to read first data
                                  inboundChannel.read();
                              } else {
                                  // Close the connection if the connection
                                  // attempt
                                  // has failed.
                                  inboundChannel.close();
                              }
                          }
                      });
  
                      channels.add(outboundChannel);
                      connected.add(host);
                      System.out.println("Connected to " + host);
                  }
              }
  
          }
  
      }
  
      @Override
      public void channelRead(final ChannelHandlerContext ctx, Object msg)
              throws Exception {
          channels.flushAndWrite(msg);
      }
  
      @Override
      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
              throws Exception {
          cause.printStackTrace();
          ctx.close();
      }
  
      static void closeOnFlush(Channel ch) {
          if (ch.isActive()) {
              ch.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(
                      ChannelFutureListener.CLOSE);
          }
      }
  
      static class TcpProxyBackendHandler extends ChannelInboundHandlerAdapter {
  
          private final Channel inboundChannel;
  
          public TcpProxyBackendHandler(Channel inboundChannel) {
              this.inboundChannel = inboundChannel;
          }
  
          @Override
          public void channelActive(ChannelHandlerContext ctx) throws Exception {
              ctx.read();
              ctx.write(Unpooled.EMPTY_BUFFER);
          }
  
          @Override
          public void channelRead(final ChannelHandlerContext ctx, Object msg)
                  throws Exception {
              inboundChannel.writeAndFlush(msg).addListener(
                      new ChannelFutureListener() {
                          @Override
                          public void operationComplete(ChannelFuture future)
                                  throws Exception {
                              if (future.isSuccess()) {
                                  ctx.channel().read();
                              } else {
                                  future.channel().close();
                              }
                          }
                      });
          }
  
          @Override
          public void channelInactive(ChannelHandlerContext ctx) throws Exception {
              TcpProxyHandler.closeOnFlush(inboundChannel);
          }
  
          @Override
          public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                  throws Exception {
              cause.printStackTrace();
              TcpProxyHandler.closeOnFlush(ctx.channel());
          }
  
      }
  
  }
  </code></pre>