How to serve 2 connections using 3 worker threads

[Amit Phatarphekar]

Hi there -

Just as a simple example, lets say I want to handle 3 simultaneous TCP
client connections using only 2 worker threads in netty, how would I do it?

With the code below, my third connection doesn't get any data from the
server - the connection just sits there.  Notice - how my worker executor
and worker count is 2.

Another question is - Does netty use CompletionService of
java.util.concurrent?  It doesnt seem to use it.  Also, I didnt see any
source code that does executor.submit or future.get
So all this has added to the confusion of how it handles and serves data to
connections that are MORE than its worker threads?


import java.net.InetSocketAddress;
import java.nio.channels.ClosedChannelException;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.string.StringEncoder;

public class SRNGServer {

  public static void main(String[] args) throws Exception {
      // Configure the server.
      ServerBootstrap bootstrap = new ServerBootstrap(
              new NioServerSocketChannelFactory(
                      Executors.newCachedThreadPool(),
                      //Executors.newCachedThreadPool()
                      Executors.newFixedThreadPool(2),2
              ));

      // Configure the pipeline factory.
      bootstrap.setPipelineFactory(new SRNGServerPipelineFactoryP());

      // Bind and start to accept incoming connections.
      bootstrap.bind(new InetSocketAddress(8080));
  }



private static class SRNGServerHandlerP extends SimpleChannelUpstreamHandler
{
  private static final Logger logger =
Logger.getLogger(SRNGServerHandlerP.class.getName());
   @Override
  public void channelConnected(
          ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {

    // Send greeting for a new connection.
   Channel ch=e.getChannel();

   System.out.printf("channelConnected with channel=[%s]%n", ch);

      ChannelFuture writeFuture=e.getChannel().write("It is " + new Date() +
" now.\r\n");

      SRNGChannelFutureListener srngcfl=new SRNGChannelFutureListener();

   System.out.printf("Registered listener=[%s] for future=[%s]%n", srngcfl,
writeFuture);

      writeFuture.addListener(srngcfl);

  }
  @Override
  public void exceptionCaught(
          ChannelHandlerContext ctx, ExceptionEvent e) {

      logger.log(
              Level.WARNING,
              "Unexpected exception from downstream.",
              e.getCause());
      if(e.getCause() instanceof ClosedChannelException){
      logger.log(Level.INFO, "****** Connection closed by client - Closing
Channel");
      }
      e.getChannel().close();
  }
}
  private static class SRNGServerPipelineFactoryP implements
ChannelPipelineFactory {
  public ChannelPipeline getPipeline() throws Exception {

      // Create a default pipeline implementation.
      ChannelPipeline pipeline = Channels.pipeline();
/*        pipeline.addLast("framer", new DelimiterBasedFrameDecoder( 8192,
Delimiters.lineDelimiter()));
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());*/

      pipeline.addLast("encoder", new StringEncoder());
      pipeline.addLast("handler", new SRNGServerHandlerP());

      return pipeline;
  }
}

private static class SRNGChannelFutureListener implements
ChannelFutureListener{
 public void operationComplete(ChannelFuture future) throws
InterruptedException{
Thread.sleep(1000*5);
Channel ch=future.getChannel();
if(ch!=null && ch.isConnected()){
ChannelFuture writeFuture=ch.write("It is " + new Date() + " now.\r\n");
//-- Add this instance as listener itself.
writeFuture.addListener(this);
}
 }
 }
}
-------------- next part --------------
An HTML attachment was scrubbed...
URL: http://lists.jboss.org/pipermail/netty-users/attachments/20110722/c40e7a36/attachment.html