title: Netty 4 - single server, multiple clients with central queue processing
body: <p>My situtation is as follows: A single "server" has a queue onto which messages (modelled as Strings for transport) from elsewhere are placed. This queue is to be consumed by all and any clients connected to the server. Each client can be seen as a kind of workflow/BPM engine ie the received String is converted to a POJO and then evaluated to see if it meets criteria and if so further processing occurs.</p>
  
  <p>I'm trying to use the server-client approach whereby the server (hosting the queue) is connected to by the clients (receing to consume).  I have a basic pipeline factory set up for each server and client:</p>
  
  <p><code>public class ServerPipelineFactory extends ChannelInitializer
  {
      @Override
      public void initChannel( SocketChannel ch ) throws Exception
      {<br>
          pipeline.addLast( "fieldLengthPrepender", new LengthFieldPrepender( 4 ) );
          pipeline.addLast( "channelEventHandler", new ClientConnectHandler() );
          pipeline.addLast(
                  "gzipMessageDeflator",
                  ZlibCodecFactory.newZlibEncoder( ZlibWrapper.GZIP )
          );
      }
  }<br>
  public class ClientConnectHandler extends SimpleChannelInboundHandler
  {
      public static final ChannelGroup allConnected = new DefaultChannelGroup(
              GlobalEventExecutor.INSTANCE );
      ....
      @Override
      public void channelActive( ChannelHandlerContext ctx ) throws Exception
      {
          LOGGER.info( "Client Connected" );
          allConnected.add( ctx.channel() );
          super.channelActive( ctx );
      }
  }</code></p>
  
  <p>The way the server is set up (fragment) is as follows:</p>
  
  <p><code>...
      EventLoopGroup bossGroup = new NioEventLoopGroup( 1 );
      EventLoopGroup workerGroup = new NioEventLoopGroup( );
      ScheduledExecutorService svc = Executors.newSingleThreadScheduledExecutor();
      try
      {
          ServerBootstrap b = new ServerBootstrap();
          b.group( bossGroup, workerGroup ).channel( NioServerSocketChannel.class )
              .childHandler( new ServerPipelineFactory() );
          Channel ch = b.bind( port ).sync().channel();
          int count = 0;
          while ( ClientConnectHandler.allConnected.isEmpty() )
          {
              .... do nothing, ie don't poll the queue
          }</code></p>
  
  <p>Client - the handler and the factory.</p>
  
  <p><code>
  @Component
  @DependsOn( { "clientSpringConfig", "demoWorkflowProcessor" } )
  public class ClientPipelineFactory extends ChannelInitializer
  {
      @Override
      public void initChannel( SocketChannel ch ) throws Exception
      {
          ChannelPipeline pipeline = ch.pipeline();
          pipeline.addLast( createInboundHandlers() );
          pipeline.addLast( createOutboundHandlers() );
          pipeline.addLast( createLogicHandlers() );
      }
      ...
      private ChannelHandler[] createLogicHandlers()
      {
          List handlers = new ArrayList&lt;>( 1 );
          ClientChannelHandler handler =
                  ( ClientChannelHandler ) AppContext.getBean( AppContext.CLIENT_HANDLER );
          handlers.add( handler );
          return handlers.toArray( new ChannelHandler[ 1 ] );
      }
  }</code></p>
  
  <p><code>public class ClientChannelHandler extends SimpleChannelInboundHandler
  {
      @Override
      protected void channelRead0( ChannelHandlerContext ctx, String msg ) throws Exception
      {
          LOGGER.info( "ADDING QUEUE: " + msg );
          rcvdQueue.put( msg );
      }<br>
  }</code></p>
  
  <p>The queue on the client side is read and processed by a separate ExecutorService instance.</p>
  
  <p>My ServerSide <code>@Component</code> message sender can successfully call to  <code>ClientConnectHandler.allConnected.writeAndFlush( buffer );</code> which will write the retrieved element from the queue to all channels - not what I want.  Can anyone help?</p>
  
  <p>I have successfully got it to work in a Client -> (send message "give me entry") -> Server -> (poll queue) -> (send response) -> Client -> (stick on client processing queue) and doing a "broadcast" from the server to client but what eludes me is the round robin without the client-inititated message.</p>
  
  <p>Thanks in advance,</p>
  
  <p>Al.</p>