- memory leak fix: TransportConnection would leak memory if an error occured while start()ing the connection. Most visible when you create a network connector pointing at a remote broker that was down since this loops through creating TransportConnectors every few seconds.

- Deadlock fix VMTransport could dead lock if during start() an error occured and stop was recusively called.



git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@592114 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java

@@ -846,6 +846,10 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
             active = true;
             this.processDispatch(connector.getBrokerInfo());
             connector.onStarted(this);
+        } catch (Exception e) {
+            // Force clean up on an error starting up.
+            stop();
+            throw e;
         } finally {
             // stop() can be called from within the above block,
             // but we want to be sure start() completes before

activemq-core/src/main/java/org/apache/activemq/thread/Valve.java

@@ -48,7 +48,7 @@ public final class Valve {
         }
     }
 
-    boolean isOn() {
+    public boolean isOn() {
         synchronized (mutex) {
             return on;
         }

activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java

@@ -115,7 +115,7 @@ public class VMTransport implements Transport, Task {
             enqueueValve.turnOff();
             if (messageQueue != null && !async) {
                 Object command;
-                while ((command = messageQueue.poll()) != null) {
+                while ((command = messageQueue.poll()) != null && !stopping.get() ) {
                     transportListener.onCommand(command);
                 }
             }
@@ -124,27 +124,37 @@ public class VMTransport implements Transport, Task {
         } finally {
             enqueueValve.turnOn();
         }
+        // If we get stopped while starting up, then do the actual stop now 
+        // that the enqueueValve is back on.
+        if( stopping.get() ) {
+            stop();
+        }
     }
 
     public void stop() throws Exception {
-        TaskRunner tr = null;
+        stopping.set(true);
-        try {
+        
-            stopping.set(true);
+        // If stop() is called while being start()ed.. then we can't stop until we return to the start() method.
-            enqueueValve.turnOff();
+        if( enqueueValve.isOn() ) {
-            if (!disposed) {
+            
-                started = false;
+            TaskRunner tr = null;
-                disposed = true;
+            try {
-                if (taskRunner != null) {
+                enqueueValve.turnOff();
-                    tr = taskRunner;
+                if (!disposed) {
-                    taskRunner = null;
+                    started = false;
+                    disposed = true;
+                    if (taskRunner != null) {
+                        tr = taskRunner;
+                        taskRunner = null;
+                    }
                 }
+            } finally {
+                stopping.set(false);
+                enqueueValve.turnOn();
+            }
+            if (tr != null) {
+                tr.shutdown(1000);
             }
-        } finally {
-            stopping.set(false);
-            enqueueValve.turnOn();
-        }
-        if (tr != null) {
-            tr.shutdown(1000);
         }
     }