https://issues.apache.org/jira/browse/AMQ-5346

need to call free on the amqp resources now to ensure their state data is all cleaned up.
yape · Sep 8, 2014 · 3afde7b · 3afde7b
1 parent 1abc95b
commit 3afde7b
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 11 deletions.
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
@@ -32,6 +32,7 @@
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQMessage;
 import org.apache.activemq.command.ActiveMQTempQueue;
+import org.apache.activemq.command.ActiveMQTempTopic;
 import org.apache.activemq.command.Command;
 import org.apache.activemq.command.ConnectionError;
 import org.apache.activemq.command.ConnectionId;
@@ -53,7 +54,6 @@
 import org.apache.activemq.command.SessionInfo;
 import org.apache.activemq.command.ShutdownInfo;
 import org.apache.activemq.command.TransactionInfo;
-import org.apache.activemq.command.ActiveMQTempTopic;
 import org.apache.activemq.selector.SelectorParser;
 import org.apache.activemq.util.IOExceptionSupport;
 import org.apache.activemq.util.IdGenerator;
@@ -83,7 +83,6 @@
 import org.apache.qpid.proton.engine.Delivery;
 import org.apache.qpid.proton.engine.EndpointState;
 import org.apache.qpid.proton.engine.Event;
-import org.apache.qpid.proton.engine.Event.Type.*;
 import org.apache.qpid.proton.engine.Link;
 import org.apache.qpid.proton.engine.Receiver;
 import org.apache.qpid.proton.engine.Sasl;
@@ -310,6 +309,7 @@ protected void processLinkEvent(Link link) throws Exception {
         } else if (remoteState == EndpointState.CLOSED) {
             ((AmqpDeliveryListener) link.getContext()).onClose();
             link.close();
+            link.free();
         }
     }
 
@@ -481,6 +481,7 @@ private void onSessionClose(Session session) {
             session.setContext(null);
         }
         session.close();
+        session.free();
     }
 
     private void onLinkOpen(Link link) {

diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
@@ -46,11 +46,9 @@
 import org.apache.activemq.broker.jmx.QueueViewMBean;
 import org.apache.activemq.transport.amqp.joram.ActiveMQAdmin;
 import org.apache.activemq.util.Wait;
-
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
-
 import org.objectweb.jtests.jms.framework.TestConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -284,7 +282,7 @@ public void testSelectors() throws Exception{
             assertEquals("hello + 9", ((TextMessage) msg).getText());
         }
     }
-    
+
     abstract class Testable implements Runnable {
         protected String msg;
         synchronized boolean passed() {
@@ -307,8 +305,9 @@ public void testProducerThrowsWhenBrokerStops() throws Exception {
         producer.setDeliveryMode(DeliveryMode.PERSISTENT);
 
         final Message m = session.createTextMessage("Sample text");
-        
+
         Testable t = new Testable() {
+            @Override
             public synchronized void run() {
                 try {
                     for (int i = 0; i < 30; ++i) {
@@ -329,7 +328,7 @@ public synchronized void run() {
             //wait until we know that the producer was able to send a message
             producer.wait(10000);
         }
-            
+
         stopBroker();
         assertTrue(t.passed());
     }
@@ -341,8 +340,9 @@ public void testProducerCreateThrowsWhenBrokerStops() throws Exception {
         final Queue queue = session.createQueue(getDestinationName());
         connection.start();
 
-        
+
         Testable t = new Testable() {
+            @Override
             public synchronized void run() {
                 try {
                     for (int i = 0; i < 10; ++i) {
@@ -400,6 +400,7 @@ public void testConsumerReceiveNoWaitThrowsWhenBrokerStops() throws Exception {
 
         final MessageConsumer consumer=session.createConsumer(queue);
         Testable t = new Testable() {
+            @Override
             public synchronized void run() {
                 try {
                     for (int i = 0; i < 10; ++i) {
@@ -414,7 +415,7 @@ public synchronized void run() {
                     LOG.info("Caught exception on receiveNoWait: {}", ex);
                 }
             }
-            
+
         };
         synchronized (consumer) {
             new Thread(t).start();
@@ -433,6 +434,7 @@ public void testConsumerReceiveTimedThrowsWhenBrokerStops() throws Exception {
 
         final MessageConsumer consumer=session.createConsumer(queue);
         Testable t = new Testable() {
+            @Override
             public synchronized void run() {
                 try {
                     for (int i = 0; i < 10; ++i) {
@@ -464,8 +466,9 @@ public void testConsumerReceiveReturnsBrokerStops() throws Exception {
         connection.start();
 
         final MessageConsumer consumer=session.createConsumer(queue);
-        
+
         Testable t = new Testable() {
+            @Override
             public synchronized void run() {
                 try {
                     Message m = consumer.receive(1);
@@ -747,7 +750,7 @@ public void onException(JMSException exception) {
         s.createTemporaryQueue().delete();
 
         stopBroker();
-        
+
         assertTrue("No exception listener event fired.", called.await(15, TimeUnit.SECONDS));
     }