http://issues.apache.org/activemq/browse/AMQ-847

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

activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java

@@ -540,10 +540,12 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
                 if (!closed.get()) {
                     closing.set(true);
 
-                    if (advisoryConsumer != null) {
-                        advisoryConsumer.dispose();
-                        advisoryConsumer = null;
-                    }
+                this.factoryStats.removeConnection(this);
+                
+                if( advisoryConsumer!=null ) {
+                    advisoryConsumer.dispose();
+                    advisoryConsumer=null;
+                }
 
                     for (Iterator i = this.sessions.iterator(); i.hasNext();) {
                         ActiveMQSession s = (ActiveMQSession) i.next();

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

@@ -389,6 +389,13 @@ public abstract class AbstractConnection implements Service, Connection, Task, C
     }
 
     public Response processBrokerInfo(BrokerInfo info) {
+    	
+    	// We only expect to get one broker info command per connection
+    	if( this.brokerInfo!=null ) {
+    		log.warn("Unexpected extra broker info command received: "+info);
+    	}
+    	
+        this.brokerInfo = info;
         broker.addBroker(this, info);
         return null;
     }

activemq-core/src/main/java/org/apache/activemq/broker/region/IndirectMessageReference.java

@@ -146,6 +146,7 @@ public class IndirectMessageReference implements MessageReference {
     
     synchronized public void drop() {
         dropped=true;
+        lockOwner = null;
         if( !persistent && message!=null ) {
             message.decrementReferenceCount();
             message=null;
@@ -156,7 +157,7 @@ public class IndirectMessageReference implements MessageReference {
         if( !regionDestination.lock(this, subscription) )
             return false;        
         synchronized (this) {
-            if( lockOwner!=null && lockOwner!=subscription )
+            if( dropped || (lockOwner!=null && lockOwner!=subscription) )
                 return false;
             lockOwner = subscription;
             return true;

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

@@ -51,6 +51,7 @@ public class Scheduler {
         ScheduledFuture ticket = (ScheduledFuture) clockTickets.remove(task);
         if( ticket!=null ) {
             ticket.cancel(true);
+        	clockDaemon.remove(task);
         }
     }
 

activemq-core/src/test/resources/org/apache/activemq/network/reconnect-broker1.xml

@@ -19,6 +19,18 @@
 
   <broker brokerName="broker1" persistent="false" useShutdownHook="false" useJmx="false"  xmlns="http://activemq.org/config/1.0">
 
+    <!-- disabling the subscription recovery policy allows us take memory leaks easier -->
+    <destinationPolicy>
+      <policyMap>
+        <policyEntries>
+          <policyEntry topic=">">
+            <subscriptionRecoveryPolicy>
+              <noSubscriptionRecoveryPolicy />
+            </subscriptionRecoveryPolicy>
+          </policyEntry>
+      </policyEntries></policyMap>
+    </destinationPolicy>
+    
     <transportConnectors>
       <transportConnector uri="tcp://localhost:61616"/>
       <transportConnector uri="vm://broker1"/>

activemq-core/src/test/resources/org/apache/activemq/network/reconnect-broker2.xml

@@ -19,6 +19,18 @@
 
   <broker brokerName="broker2" persistent="false" useShutdownHook="false" useJmx="false" xmlns="http://activemq.org/config/1.0">
 
+    <!-- disabling the subscription recovery policy allows us take memory leaks easier -->
+    <destinationPolicy>
+      <policyMap>
+        <policyEntries>
+          <policyEntry topic=">">
+            <subscriptionRecoveryPolicy>
+              <noSubscriptionRecoveryPolicy />
+            </subscriptionRecoveryPolicy>
+          </policyEntry>
+      </policyEntries></policyMap>
+    </destinationPolicy>
+
     <transportConnectors>
       <transportConnector uri="tcp://localhost:61617"/>
       <transportConnector uri="vm://broker2"/>

activemq-core/src/test/resources/org/apache/activemq/network/ssh-reconnect-broker1.xml

@@ -19,6 +19,18 @@
 
   <broker brokerName="broker1" persistent="false" useShutdownHook="false" useJmx="false" xmlns="http://activemq.org/config/1.0">
 
+    <!-- disabling the subscription recovery policy allows us take memory leaks easier -->
+    <destinationPolicy>
+      <policyMap>
+        <policyEntries>
+          <policyEntry topic=">">
+            <subscriptionRecoveryPolicy>
+              <noSubscriptionRecoveryPolicy />
+            </subscriptionRecoveryPolicy>
+          </policyEntry>
+      </policyEntries></policyMap>
+    </destinationPolicy>
+    
     <transportConnectors>
       <transportConnector uri="tcp://localhost:61616"/>
       <transportConnector uri="vm://broker1"/>

activemq-core/src/test/resources/org/apache/activemq/network/ssh-reconnect-broker2.xml

@@ -19,6 +19,18 @@
 
   <broker brokerName="broker2" persistent="false" useShutdownHook="false" useJmx="false" xmlns="http://activemq.org/config/1.0">
 
+    <!-- disabling the subscription recovery policy allows us take memory leaks easier -->
+    <destinationPolicy>
+      <policyMap>
+        <policyEntries>
+          <policyEntry topic=">">
+            <subscriptionRecoveryPolicy>
+              <noSubscriptionRecoveryPolicy />
+            </subscriptionRecoveryPolicy>
+          </policyEntry>
+      </policyEntries></policyMap>
+    </destinationPolicy>
+    
     <transportConnectors>
       <transportConnector uri="tcp://localhost:61617"/>
       <transportConnector uri="vm://broker2"/>