[jboss-remoting-commits] JBoss Remoting SVN: r4414 - remoting3/trunk/protocol/basic/src/test/java/org/jboss/cx/remoting/protocol/basic.

[jboss-remoting-commits at lists.jboss.org]

Author: david.lloyd at jboss.com
Date: 2008-07-21 09:01:14 -0400 (Mon, 21 Jul 2008)
New Revision: 4414

Modified:
   remoting3/trunk/protocol/basic/src/test/java/org/jboss/cx/remoting/protocol/basic/ConnectionTestCase.java
Log:
Use new XNIO methods for clean shutdown of executor, channel source

Modified: remoting3/trunk/protocol/basic/src/test/java/org/jboss/cx/remoting/protocol/basic/ConnectionTestCase.java
===================================================================
--- remoting3/trunk/protocol/basic/src/test/java/org/jboss/cx/remoting/protocol/basic/ConnectionTestCase.java	2008-07-20 21:10:21 UTC (rev 4413)
+++ remoting3/trunk/protocol/basic/src/test/java/org/jboss/cx/remoting/protocol/basic/ConnectionTestCase.java	2008-07-21 13:01:14 UTC (rev 4414)
@@ -49,9 +49,10 @@
 import org.jboss.xnio.IoHandlerFactory;
 import org.jboss.xnio.ConfigurableFactory;
 import org.jboss.xnio.IoFuture;
-import org.jboss.xnio.TcpConnector;
 import org.jboss.xnio.TcpClient;
 import org.jboss.xnio.ChannelSource;
+import org.jboss.xnio.CloseableTcpConnector;
+import org.jboss.xnio.CloseableExecutor;
 import org.jboss.xnio.channels.AllocatedMessageChannel;
 import org.jboss.xnio.channels.Channels;
 import org.jboss.xnio.channels.StreamChannel;
@@ -68,7 +69,7 @@
         final String REQUEST = "request";
         final String REPLY = "reply";
         final List<Throwable> problems = Collections.synchronizedList(new LinkedList<Throwable>());
-        final ExecutorService executorService = Executors.newCachedThreadPool();
+        final CloseableExecutor closeableExecutor = IoUtils.closeableExecutor(Executors.newCachedThreadPool(), 500L, TimeUnit.MILLISECONDS);
         try {
             final BufferAllocator<ByteBuffer> allocator = new BufferAllocator<ByteBuffer>() {
                 public ByteBuffer allocate() {
@@ -81,7 +82,7 @@
             final Xnio xnio = Xnio.createNio();
             try {
                 final EndpointImpl endpoint = new EndpointImpl();
-                endpoint.setExecutor(executorService);
+                endpoint.setExecutor(closeableExecutor);
                 endpoint.start();
                 try {
                     final ServiceRegistry serviceRegistry = new ServiceRegistryImpl();
@@ -97,17 +98,17 @@
                         });
                         try {
                             serviceRegistry.bind(serviceEndpointHandle.getResource(), 13);
-                            final IoHandlerFactory<AllocatedMessageChannel> handlerFactory = BasicProtocol.createServer(executorService, allocator, serviceRegistry);
+                            final IoHandlerFactory<AllocatedMessageChannel> handlerFactory = BasicProtocol.createServer(closeableExecutor, allocator, serviceRegistry);
                             final IoHandlerFactory<StreamChannel> newHandlerFactory = Channels.convertStreamToAllocatedMessage(handlerFactory, 32768, 32768);
                             final ConfigurableFactory<Closeable> tcpServerFactory = xnio.createTcpServer(newHandlerFactory, new InetSocketAddress(12345));
                             final Closeable tcpServerCloseable = tcpServerFactory.create();
                             try {
-                                final ConfigurableFactory<TcpConnector> connectorFactory = xnio.createTcpConnector();
-                                final TcpConnector connector = connectorFactory.create();
+                                final ConfigurableFactory<CloseableTcpConnector> connectorFactory = xnio.createTcpConnector();
+                                final CloseableTcpConnector connector = connectorFactory.create();
                                 try {
                                     final TcpClient tcpClient = connector.createChannelSource(new InetSocketAddress("localhost", 12345));
                                     final ChannelSource<AllocatedMessageChannel> channelSource = Channels.convertStreamToAllocatedMessage(tcpClient, 32768, 32768);
-                                    final IoFuture<Connection> futureCloseable = BasicProtocol.connect(executorService, channelSource, allocator, serviceRegistry);
+                                    final IoFuture<Connection> futureCloseable = BasicProtocol.connect(closeableExecutor, channelSource, allocator, serviceRegistry);
                                     final Connection connection = futureCloseable.get();
                                     try {
                                         final Handle<RemoteServiceEndpoint> handleThirteen = connection.getServiceForId(13);
@@ -132,7 +133,7 @@
                                         IoUtils.safeClose(connection);
                                     }
                                 } finally {
-                                    // todo close connector
+                                    IoUtils.safeClose(connector);
                                 }
                             } finally {
                                 IoUtils.safeClose(tcpServerCloseable);
@@ -150,7 +151,7 @@
                 IoUtils.safeClose(xnio);
             }
         } finally {
-            executorService.shutdownNow();
+            IoUtils.safeClose(closeableExecutor);
         }
         for (Throwable t : problems) {
             throw t;