Hi there -<div><br></div><div>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?</div><div><br></div><div>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.</div>
<div><br></div><div>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</div><div>So all this has added to the confusion of how it handles and serves data to connections that are MORE than its worker threads?</div>
<div><br></div><div><br></div><div><div>import java.net.InetSocketAddress;</div><div>import java.nio.channels.ClosedChannelException;</div><div>import java.util.Date;</div><div>import java.util.concurrent.Executors;</div>
<div>import java.util.logging.Level;</div><div>import java.util.logging.Logger;</div><div><br></div><div>import org.jboss.netty.bootstrap.ServerBootstrap;</div><div>import org.jboss.netty.channel.Channel;</div><div>import org.jboss.netty.channel.ChannelFuture;</div>
<div>import org.jboss.netty.channel.ChannelFutureListener;</div><div>import org.jboss.netty.channel.ChannelHandlerContext;</div><div>import org.jboss.netty.channel.ChannelPipeline;</div><div>import org.jboss.netty.channel.ChannelPipelineFactory;</div>
<div>import org.jboss.netty.channel.ChannelStateEvent;</div><div>import org.jboss.netty.channel.Channels;</div><div>import org.jboss.netty.channel.ExceptionEvent;</div><div>import org.jboss.netty.channel.SimpleChannelUpstreamHandler;</div>
<div>import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;</div><div>import org.jboss.netty.handler.codec.string.StringEncoder;</div><div><br></div><div>public class SRNGServer {</div><div><br></div><div>
public static void main(String[] args) throws Exception {</div><div> // Configure the server.</div><div> ServerBootstrap bootstrap = new ServerBootstrap(</div><div> new NioServerSocketChannelFactory(</div>
<div> Executors.newCachedThreadPool(),</div><div> //Executors.newCachedThreadPool()</div><div> Executors.newFixedThreadPool(2),2</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>));</div>
<div><br></div><div> // Configure the pipeline factory.</div><div> bootstrap.setPipelineFactory(new SRNGServerPipelineFactoryP());</div><div><br></div><div> // Bind and start to accept incoming connections.</div>
<div> bootstrap.bind(new InetSocketAddress(8080));</div><div> }</div><div><br></div><div><br></div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>private static class SRNGServerHandlerP extends SimpleChannelUpstreamHandler {</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span> private static final Logger logger = Logger.getLogger(SRNGServerHandlerP.class.getName());</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span> @Override</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> public void channelConnected(</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> <span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span> // Send greeting for a new connection.</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> <span class="Apple-tab-span" style="white-space:pre"> </span>Channel ch=e.getChannel();</div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span> <span class="Apple-tab-span" style="white-space:pre"> </span>System.out.printf("channelConnected with channel=[%s]%n", ch);</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> <span class="Apple-tab-span" style="white-space:pre"> </span></div><div> ChannelFuture writeFuture=e.getChannel().write("It is " + new Date() + " now.\r\n");</div>
<div> </div><div> SRNGChannelFutureListener srngcfl=new SRNGChannelFutureListener();</div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span> <span class="Apple-tab-span" style="white-space:pre"> </span>System.out.printf("Registered listener=[%s] for future=[%s]%n", srngcfl, writeFuture);</div>
<div> </div><div> writeFuture.addListener(srngcfl); </div><div><span class="Apple-tab-span" style="white-space:pre"> </span> </div><div><span class="Apple-tab-span" style="white-space:pre"> </span> }</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span> @Override</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> public void exceptionCaught(</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> ChannelHandlerContext ctx, ExceptionEvent e) {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> <span class="Apple-tab-span" style="white-space:pre"> </span></div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> logger.log(</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> Level.WARNING,</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> "Unexpected exception from downstream.",</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> e.getCause());</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> if(e.getCause() instanceof ClosedChannelException){</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> logger.log(Level.INFO, "****** Connection closed by client - Closing Channel");</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> }</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> e.getChannel().close();</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> }</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div>
<span class="Apple-tab-span" style="white-space:pre"> </span>private static class SRNGServerPipelineFactoryP implements ChannelPipelineFactory {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span></div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> public ChannelPipeline getPipeline() throws Exception {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> <span class="Apple-tab-span" style="white-space:pre"> </span></div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span> // Create a default pipeline implementation.</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> ChannelPipeline pipeline = Channels.pipeline();</div>
<div>/* pipeline.addLast("framer", new DelimiterBasedFrameDecoder( 8192, Delimiters.lineDelimiter()));</div><div> pipeline.addLast("decoder", new StringDecoder());</div><div> pipeline.addLast("encoder", new StringEncoder());*/</div>
<div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span> pipeline.addLast("encoder", new StringEncoder());</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> pipeline.addLast("handler", new SRNGServerHandlerP());</div>
<div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span> return pipeline;</div><div><span class="Apple-tab-span" style="white-space:pre"> </span> }</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>private static class SRNGChannelFutureListener implements ChannelFutureListener{</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>public void operationComplete(ChannelFuture future) throws InterruptedException{</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>Thread.sleep(1000*5);</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>Channel ch=future.getChannel();</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>if(ch!=null && ch.isConnected()){</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>ChannelFuture writeFuture=ch.write("It is " + new Date() + " now.\r\n");</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>//-- Add this instance as listener itself.</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>writeFuture.addListener(this);</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div><span class="Apple-tab-span" style="white-space:pre"> </span></div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div>
}</div></div><div><br></div>