4:54 a.m.
Author: ataylor
Date: 2011-10-13 05:54:50 -0400 (Thu, 13 Oct 2011)
New Revision: 11529
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
trunk/hornetq-core/src/test/java/org/hornetq/tests/util/UnitTestCase.java
Log:
added check for hanging retries after shutdown
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-10-13
07:06:03 UTC (rev 11528)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-10-13
09:54:50 UTC (rev 11529)
@@ -16,15 +16,8 @@
import java.lang.ref.WeakReference;
import java.security.AccessController;
import java.security.PrivilegedAction;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.*;
+import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import org.hornetq.api.core.HornetQBuffer;
@@ -156,6 +149,8 @@
private final Object waitLock = new Object();
+ public final static List<CloseRunnable> CLOSE_RUNNABLES = new
ArrayList<CloseRunnable>();
+
// Static
//
---------------------------------------------------------------------------------------
@@ -1453,17 +1448,7 @@
serverLocator.notifyNodeDown(System.currentTimeMillis(),
msg.getNodeID().toString());
}
- closeExecutor.execute(new Runnable()
- {
- // Must be executed on new thread since cannot block the netty thread for
a long time and fail can
- // cause reconnect loop
- public void run()
- {
- conn.fail(new HornetQException(HornetQException.DISCONNECTED,
- "The connection was disconnected
because of server shutdown"));
-
- }
- });
+ closeExecutor.execute(new CloseRunnable(conn));
}
else if (type == PacketImpl.CLUSTER_TOPOLOGY)
{
@@ -1520,8 +1505,42 @@
}
}
}
+
+
}
+ public class CloseRunnable implements Runnable
+ {
+ private CoreRemotingConnection conn;
+
+ public CloseRunnable(CoreRemotingConnection conn)
+ {
+ this.conn = conn;
+ }
+
+ // Must be executed on new thread since cannot block the netty thread for a long
time and fail can
+ // cause reconnect loop
+ public void run()
+ {
+ CLOSE_RUNNABLES.add(this);
+ try
+ {
+ conn.fail(new HornetQException(HornetQException.DISCONNECTED,
+ "The connection was disconnected because of server
shutdown"));
+ } finally
+ {
+ CLOSE_RUNNABLES.remove(this);
+ }
+
+ }
+
+ public ClientSessionFactoryImpl stop()
+ {
+ causeExit();
+ return ClientSessionFactoryImpl.this;
+ }
+
+ }
private class DelegatingBufferHandler implements BufferHandler
{
public void bufferReceived(final Object connectionID, final HornetQBuffer buffer)
Modified: trunk/hornetq-core/src/test/java/org/hornetq/tests/util/UnitTestCase.java
===================================================================
--- trunk/hornetq-core/src/test/java/org/hornetq/tests/util/UnitTestCase.java 2011-10-13
07:06:03 UTC (rev 11528)
+++ trunk/hornetq-core/src/test/java/org/hornetq/tests/util/UnitTestCase.java 2011-10-13
09:54:50 UTC (rev 11529)
@@ -57,6 +57,7 @@
import org.hornetq.api.core.client.ClientMessage;
import org.hornetq.api.core.client.ClientSession;
import org.hornetq.core.asyncio.impl.AsynchronousFileImpl;
+import org.hornetq.core.client.impl.ClientSessionFactoryImpl;
import org.hornetq.core.client.impl.ServerLocatorImpl;
import org.hornetq.core.config.Configuration;
import org.hornetq.core.config.impl.ConfigurationImpl;
@@ -913,8 +914,25 @@
@Override
protected void tearDown() throws Exception
{
+ List<ClientSessionFactoryImpl.CloseRunnable> closeRunnables = new
ArrayList<ClientSessionFactoryImpl.CloseRunnable>(ClientSessionFactoryImpl.CLOSE_RUNNABLES);
+ ArrayList<Exception> exceptions = new ArrayList<Exception>();
+ if(!closeRunnables.isEmpty())
+ {
+ for (ClientSessionFactoryImpl.CloseRunnable closeRunnable : closeRunnables)
+ {
+ exceptions.add(closeRunnable.stop().e);
+ }
+ }
cleanupPools();
-
+ //clean up pools before failing
+ if(!exceptions.isEmpty())
+ {
+ for (Exception exception : exceptions)
+ {
+ exception.printStackTrace();
+ }
+ fail("Client Session Factories still tryint to reconnect, see above to see
where created");
+ }
Map<Thread, StackTraceElement[]> threadMap = Thread.getAllStackTraces();
for (Thread thread : threadMap.keySet())
{
@@ -925,12 +943,12 @@
{
alreadyFailedThread.add(thread);
System.out.println(threadDump(this.getName() + " has left threads
running. Look at thread " +
- thread.getName() +
- " id = " +
- thread.getId() +
- " has running locators on test "
+
- this.getName() +
- " on this following dump"));
+ thread.getName() +
+ " id = " +
+ thread.getId() +
+ " has running locators on test " +
+ this.getName() +
+ " on this following dump"));
fail("test left serverlocator running, this could effect other
tests");
}
else if
(stackTraceElement.getMethodName().contains("BroadcastGroupImpl.run") &&
!alreadyFailedThread.contains(thread))