fix for: https://issues.apache.org/jira/browse/AMQ-3917

Allow multiple durable subs for Stomp v1.1 client connections.

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

activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java

@@ -22,10 +22,7 @@ import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.OutputStreamWriter;
 import java.io.PrintWriter;
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -603,9 +600,13 @@ public class ProtocolConverter {
 
         // check if it is a durable subscription
         String durable = command.getHeaders().get("activemq.subscriptionName");
+        String clientId = durable;
+        if (this.version.equals(Stomp.V1_1)) {
+            clientId = connectionInfo.getClientId();
+        }
         if (durable != null) {
             RemoveSubscriptionInfo info = new RemoveSubscriptionInfo();
-            info.setClientId(durable);
+            info.setClientId(clientId);
             info.setSubscriptionName(durable);
             info.setConnectionId(connectionId);
             sendToActiveMQ(info, createResponseHandler(command));

activemq-core/src/test/java/org/apache/activemq/transport/stomp/Stomp11Test.java

@@ -30,11 +30,13 @@ import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
+import javax.management.ObjectName;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.CombinationTestSupport;
 import org.apache.activemq.broker.BrokerFactory;
 import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.jmx.BrokerViewMBean;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ActiveMQTextMessage;
 import org.slf4j.Logger;
@@ -638,11 +640,11 @@ public class Stomp11Test extends CombinationTestSupport {
         assertTrue(frame.startsWith("CONNECTED"));
 
         frame = "SEND\n" +
-        		"value:newest" + "\n" +
+                "value:newest" + "\n" +
-        		"value:older" + "\n" +
+                "value:older" + "\n" +
-        		"value:oldest" + "\n" +
+                "value:oldest" + "\n" +
-        		"destination:/queue/" + getQueueName() +
+                "destination:/queue/" + getQueueName() +
-        		"\n\n" + "Hello World" + Stomp.NULL;
+                "\n\n" + "Hello World" + Stomp.NULL;
 
         stompConnection.sendFrame(frame);
 
@@ -779,4 +781,209 @@ public class Stomp11Test extends CombinationTestSupport {
         frame = "DISCONNECT\n" + "\n\n" + Stomp.NULL;
         stompConnection.sendFrame(frame);
     }
+
+    public void testDurableSubAndUnSubOnTwoTopics() throws Exception {
+        stompConnection.setVersion(Stomp.V1_1);
+
+        String domain = "org.apache.activemq";
+        ObjectName brokerName = new ObjectName(domain + ":Type=Broker,BrokerName=localhost");
+
+        BrokerViewMBean view = (BrokerViewMBean)broker.getManagementContext().newProxyInstance(brokerName, BrokerViewMBean.class, true);
+
+        String connectFrame = "STOMP\n" +
+                "login:system\n" + "passcode:manager\n" + "accept-version:1.1\n" +
+                "host:localhost\n" + "client-id:test\n" + "\n" + Stomp.NULL;
+        stompConnection.sendFrame(connectFrame);
+
+        String frame = stompConnection.receiveFrame();
+        LOG.debug("Broker sent: " + frame);
+
+        assertTrue(frame.startsWith("CONNECTED"));
+        assertEquals(view.getDurableTopicSubscribers().length, 0);
+
+        // subscribe to first destination durably
+        frame = "SUBSCRIBE\n" +
+                "destination:/topic/" + getQueueName() + "1" + "\n" +
+                "ack:auto\n" + "receipt:1\n" + "id:durablesub-1\n" +
+                "activemq.subscriptionName:test1\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+
+        StompFrame receipt = stompConnection.receive();
+        LOG.debug("Broker sent: " + receipt);
+        assertTrue(receipt.getAction().startsWith("RECEIPT"));
+        assertEquals("1", receipt.getHeaders().get("receipt-id"));
+        assertEquals(view.getDurableTopicSubscribers().length, 1);
+
+        // subscribe to second destination durably
+        frame = "SUBSCRIBE\n" +
+                "destination:/topic/" + getQueueName() + "2" + "\n" +
+                "ack:auto\n" + "receipt:2\n" + "id:durablesub-2\n" +
+                "activemq.subscriptionName:test2\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+
+        receipt = stompConnection.receive();
+        LOG.debug("Broker sent: " + receipt);
+        assertTrue(receipt.getAction().startsWith("RECEIPT"));
+        assertEquals("2", receipt.getHeaders().get("receipt-id"));
+        assertEquals(view.getDurableTopicSubscribers().length, 2);
+
+        frame = "DISCONNECT\nclient-id:test\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+        try {
+            Thread.sleep(400);
+        } catch (InterruptedException e){}
+
+        // reconnect and send some messages to the offline subscribers and then try to get
+        // them after subscribing again.
+        stompConnect();
+        stompConnection.sendFrame(connectFrame);
+        frame = stompConnection.receiveFrame();
+        LOG.debug("Broker sent: " + frame);
+        assertTrue(frame.startsWith("CONNECTED"));
+        assertEquals(view.getDurableTopicSubscribers().length, 0);
+        assertEquals(view.getInactiveDurableTopicSubscribers().length, 2);
+
+        // unsubscribe from topic 1
+        frame = "UNSUBSCRIBE\n" + "destination:/topic/" + getQueueName() + "1\n" +
+                "id:durablesub-1\n" + "receipt:3\n" +
+                "activemq.subscriptionName:test1\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+        receipt = stompConnection.receive();
+        LOG.debug("Broker sent: " + frame);
+        assertTrue(receipt.getAction().startsWith("RECEIPT"));
+        assertEquals("3", receipt.getHeaders().get("receipt-id"));
+
+        assertEquals(view.getInactiveDurableTopicSubscribers().length, 1);
+
+        // unsubscribe from topic 2
+        frame = "UNSUBSCRIBE\n" + "destination:/topic/" + getQueueName() + "2\n" +
+                "id:durablesub-2\n" + "receipt:4\n" +
+                "activemq.subscriptionName:test2\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+        receipt = stompConnection.receive();
+        LOG.debug("Broker sent: " + frame);
+        assertTrue(receipt.getAction().startsWith("RECEIPT"));
+        assertEquals("4", receipt.getHeaders().get("receipt-id"));
+
+        assertEquals(view.getInactiveDurableTopicSubscribers().length, 0);
+
+        frame = "DISCONNECT\n" + "\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+    }
+
+    public void testMultipleDurableSubsWithOfflineMessages() throws Exception {
+        stompConnection.setVersion(Stomp.V1_1);
+
+        String domain = "org.apache.activemq";
+        ObjectName brokerName = new ObjectName(domain + ":Type=Broker,BrokerName=localhost");
+
+        BrokerViewMBean view = (BrokerViewMBean)broker.getManagementContext().newProxyInstance(brokerName, BrokerViewMBean.class, true);
+
+        String connectFrame = "STOMP\n" + "login:system\n" + "passcode:manager\n" +
+                "accept-version:1.1\n" + "host:localhost\n" + "client-id:test\n" + "\n" + Stomp.NULL;
+        stompConnection.sendFrame(connectFrame);
+
+        String frame = stompConnection.receiveFrame();
+        LOG.debug("Broker sent: " + frame);
+
+        assertTrue(frame.startsWith("CONNECTED"));
+        assertEquals(view.getDurableTopicSubscribers().length, 0);
+
+        // subscribe to first destination durably
+        frame = "SUBSCRIBE\n" +
+                "destination:/topic/" + getQueueName() + "1" + "\n" +
+                "ack:auto\n" + "receipt:1\n" + "id:durablesub-1\n" +
+                "activemq.subscriptionName:test1\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+
+        StompFrame receipt = stompConnection.receive();
+        LOG.debug("Broker sent: " + receipt);
+        assertTrue(receipt.getAction().startsWith("RECEIPT"));
+        assertEquals("1", receipt.getHeaders().get("receipt-id"));
+        assertEquals(view.getDurableTopicSubscribers().length, 1);
+
+        // subscribe to second destination durably
+        frame = "SUBSCRIBE\n" +
+                "destination:/topic/" + getQueueName() + "2" + "\n" +
+                "ack:auto\n" + "receipt:2\n" + "id:durablesub-2\n" +
+                "activemq.subscriptionName:test2\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+
+        receipt = stompConnection.receive();
+        LOG.debug("Broker sent: " + receipt);
+        assertTrue(receipt.getAction().startsWith("RECEIPT"));
+        assertEquals("2", receipt.getHeaders().get("receipt-id"));
+        assertEquals(view.getDurableTopicSubscribers().length, 2);
+
+        frame = "DISCONNECT\nclient-id:test\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+        try {
+            Thread.sleep(400);
+        } catch (InterruptedException e){}
+
+        // reconnect and send some messages to the offline subscribers and then try to get
+        // them after subscribing again.
+        stompConnect();
+        stompConnection.sendFrame(connectFrame);
+        frame = stompConnection.receiveFrame();
+        LOG.debug("Broker sent: " + frame);
+        assertTrue(frame.startsWith("CONNECTED"));
+        assertEquals(view.getDurableTopicSubscribers().length, 0);
+        assertEquals(view.getInactiveDurableTopicSubscribers().length, 2);
+
+        frame = "SEND\n" + "destination:/topic/" + getQueueName() + "1\n" +
+                "receipt:10\n" + "\n" + "Hello World 1" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+        receipt = stompConnection.receive();
+        assertEquals("10", receipt.getHeaders().get(Stomp.Headers.Response.RECEIPT_ID));
+
+        frame = "SEND\n" + "destination:/topic/" + getQueueName() + "2\n" +
+                "receipt:11\n" + "\n" + "Hello World 2" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+        receipt = stompConnection.receive();
+        assertEquals("11", receipt.getHeaders().get(Stomp.Headers.Response.RECEIPT_ID));
+
+        // subscribe to first destination durably
+        frame = "SUBSCRIBE\n" +
+                "destination:/topic/" + getQueueName() + "1" + "\n" +
+                "ack:auto\n" + "receipt:3\n" + "id:durablesub-1\n" +
+                "activemq.subscriptionName:test1\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+
+        receipt = stompConnection.receive();
+        LOG.debug("Broker sent: " + receipt);
+        assertTrue(receipt.getAction().startsWith("RECEIPT"));
+        assertEquals("3", receipt.getHeaders().get("receipt-id"));
+        assertEquals(view.getDurableTopicSubscribers().length, 1);
+
+        StompFrame message = stompConnection.receive();
+        assertEquals(Stomp.Responses.MESSAGE, message.getAction());
+        assertEquals("durablesub-1", message.getHeaders().get(Stomp.Headers.Message.SUBSCRIPTION));
+
+        assertEquals(view.getDurableTopicSubscribers().length, 1);
+        assertEquals(view.getInactiveDurableTopicSubscribers().length, 1);
+
+        // subscribe to second destination durably
+        frame = "SUBSCRIBE\n" +
+                "destination:/topic/" + getQueueName() + "2" + "\n" +
+                "ack:auto\n" + "receipt:4\n" + "id:durablesub-2\n" +
+                "activemq.subscriptionName:test2\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+
+        receipt = stompConnection.receive();
+        LOG.debug("Broker sent: " + receipt);
+        assertTrue(receipt.getAction().startsWith("RECEIPT"));
+        assertEquals("4", receipt.getHeaders().get("receipt-id"));
+        assertEquals(view.getDurableTopicSubscribers().length, 2);
+
+        message = stompConnection.receive();
+        assertEquals(Stomp.Responses.MESSAGE, message.getAction());
+        assertEquals("durablesub-2", message.getHeaders().get(Stomp.Headers.Message.SUBSCRIPTION));
+
+        assertEquals(view.getDurableTopicSubscribers().length, 2);
+        assertEquals(view.getInactiveDurableTopicSubscribers().length, 0);
+
+        frame = "DISCONNECT\n" + "\n\n" + Stomp.NULL;
+        stompConnection.sendFrame(frame);
+    }
 }