[hornetq-commits] JBoss hornetq SVN: r8103 - in trunk: src/main/org/hornetq/jms/client and 1 other directories.

Author: timfox Date: 2009-10-14 05:31:42 -0400 (Wed, 14 Oct 2009) New Revision: 8103 Modified: trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java trunk/src/main/org/hornetq/jms/client/HornetQConnection.java trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java Log: a couple of tweaks Modified: trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java 2009-10-14 09:13:40 UTC (rev 8102) +++ trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java 2009-10-14 09:31:42 UTC (rev 8103) @@ -128,12 +128,11 @@ private ConcurrentMap<Object, Connection> connections = new ConcurrentHashMap<Object, Connection>(); private final String servletPath; - + private final VirtualExecutorService virtualExecutor; private ScheduledExecutorService scheduledThreadPool; - // Static -------------------------------------------------------- // Constructors -------------------------------------------------- @@ -224,10 +223,10 @@ this.tcpReceiveBufferSize = ConfigurationHelper.getIntProperty(TransportConstants.TCP_RECEIVEBUFFER_SIZE_PROPNAME, TransportConstants.DEFAULT_TCP_RECEIVEBUFFER_SIZE, configuration); - - virtualExecutor = new VirtualExecutorService(threadPool); - - this.scheduledThreadPool = scheduledThreadPool; + + virtualExecutor = new VirtualExecutorService(threadPool); + + this.scheduledThreadPool = scheduledThreadPool; } public synchronized void start() @@ -235,10 +234,10 @@ if (channelFactory != null) { return; - } - + } + if (useNio) - { + { channelFactory = new NioClientSocketChannelFactory(virtualExecutor, virtualExecutor); } else @@ -287,7 +286,7 @@ context = null; // Unused } - if(context != null && useServlet) + if (context != null && useServlet) { bootstrap.setOption("sslContext", context); } @@ -312,9 +311,11 @@ return pipeline; } }); - if(!Version.ID.equals(VersionLoader.getVersion().getNettyVersion())) + if (!Version.ID.equals(VersionLoader.getVersion().getNettyVersion())) { - log.warn("Unexpected Netty Version was expecting " + VersionLoader.getVersion().getNettyVersion() + " using " + Version.ID); + log.warn("Unexpected Netty Version was expecting " + VersionLoader.getVersion().getNettyVersion() + + " using " + + Version.ID); } log.debug("Started Netty Connector version " + Version.ID); } @@ -366,7 +367,7 @@ } } address = new InetSocketAddress(host, port); - + ChannelFuture future = bootstrap.connect(address); future.awaitUninterruptibly(); @@ -408,12 +409,12 @@ else { Throwable t = future.getCause(); - + if (t != null && !(t instanceof ConnectException)) { log.error("Failed to create netty connection", future.getCause()); } - + return null; } } @@ -469,7 +470,10 @@ if (httpClientIdleScanPeriod > 0) { task = new HttpIdleTimer(); - java.util.concurrent.Future<?> future = scheduledThreadPool.scheduleAtFixedRate(task, httpClientIdleScanPeriod, httpClientIdleScanPeriod, TimeUnit.MILLISECONDS); + java.util.concurrent.Future<?> future = scheduledThreadPool.scheduleAtFixedRate(task, + httpClientIdleScanPeriod, + httpClientIdleScanPeriod, + TimeUnit.MILLISECONDS); task.setFuture(future); } } @@ -548,6 +552,7 @@ private class HttpIdleTimer implements Runnable { private boolean closed = false; + private java.util.concurrent.Future<?> future; public synchronized void run() @@ -556,7 +561,7 @@ { return; } - + if (!waitingGet && System.currentTimeMillis() > lastSendTime + httpMaxClientIdleTime) { HttpRequest httpRequest = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, url); @@ -569,15 +574,15 @@ { this.future = future; } - + public void close() { if (future != null) - { + { future.cancel(false); } - - closed = true; + + closed = true; } } } Modified: trunk/src/main/org/hornetq/jms/client/HornetQConnection.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQConnection.java 2009-10-14 09:13:40 UTC (rev 8102) +++ trunk/src/main/org/hornetq/jms/client/HornetQConnection.java 2009-10-14 09:31:42 UTC (rev 8103) @@ -243,17 +243,15 @@ public synchronized void close() throws JMSException { - log.info("Closing jms connection"); if (closed) { - log.info("Already closed"); return; } try { for (HornetQSession session : new HashSet<HornetQSession>(sessions)) - { + { session.close(); } @@ -279,8 +277,7 @@ finally { if (initialSession != null) - { - log.info("closing initial session"); + { initialSession.close(); } } Modified: trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java 2009-10-14 09:13:40 UTC (rev 8102) +++ trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java 2009-10-14 09:31:42 UTC (rev 8103) @@ -84,9 +84,9 @@ private CountDownLatch reconnectionLatch = new CountDownLatch(1); - private boolean afterRestart = false; + private volatile boolean afterRestart = false; - private boolean receivedMessagesAfterRestart = false; + private volatile boolean receivedMessagesAfterRestart = false; private int callTimeout; @@ -98,7 +98,7 @@ { receivedMessagesAfterRestart = true; } - System.out.println(receivedMessagesAfterRestart + " " + msg); + log.info(receivedMessagesAfterRestart + " " + msg); } };