From 144dda0308848e0ae4d7f87fb229e5ff7d107fd0 Mon Sep 17 00:00:00 2001
From: Robert Davies <rajdavies@apache.org>
Date: Wed, 23 May 2012 10:25:34 +0000
Subject: [PATCH] Additional fixes for
 https://issues.apache.org/jira/browse/AMQ-3855 - timing issue in adding
 wildcard subscriptions can result in duplicate messages sent to MQTT

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1341820 13f79535-47bb-0310-9956-ffa450edef68
---
 .../apache/activemq/broker/region/Topic.java  | 24 ++++++++++++-------
 .../activemq/transport/mqtt/MQTTTest.java     |  2 +-
 2 files changed, 17 insertions(+), 9 deletions(-)

diff --git a/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java b/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
index 86b2632ee0..84c9c9d838 100755
--- a/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
+++ b/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
@@ -109,9 +109,6 @@ public class Topic extends BaseDestination implements Task {
     }
 
     public void addSubscription(ConnectionContext context, final Subscription sub) throws Exception {
-
-       super.addSubscription(context, sub);
-
         if (!sub.getConsumerInfo().isDurable()) {
 
             // Do a retroactive recovery if needed.
@@ -121,23 +118,34 @@ public class Topic extends BaseDestination implements Task {
                 // while we are recovering a subscription to avoid out of order messages.
                 dispatchLock.writeLock().lock();
                 try {
+                    boolean applyRecovery = false;
                     synchronized (consumers) {
-                        sub.add(context, this);
-                        consumers.add(sub);
+                        if (!consumers.contains(sub)){
+                            sub.add(context, this);
+                            consumers.add(sub);
+                            applyRecovery=true;
+                            super.addSubscription(context, sub);
+                        }
+                    }
+                    if (applyRecovery){
+                        subscriptionRecoveryPolicy.recover(context, this, sub);
                     }
-                    subscriptionRecoveryPolicy.recover(context, this, sub);
                 } finally {
                     dispatchLock.writeLock().unlock();
                 }
 
             } else {
                 synchronized (consumers) {
-                    sub.add(context, this);
-                    consumers.add(sub);
+                    if (!consumers.contains(sub)){
+                        sub.add(context, this);
+                        consumers.add(sub);
+                        super.addSubscription(context, sub);
+                    }
                 }
             }
         } else {
             DurableTopicSubscription dsub = (DurableTopicSubscription) sub;
+            super.addSubscription(context, sub);
     		sub.add(context, this);
     		if(dsub.isActive()) {
 	        	synchronized (consumers) {
diff --git a/activemq-core/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java b/activemq-core/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
index ce10a6f2c0..cd969a6747 100644
--- a/activemq-core/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
+++ b/activemq-core/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
@@ -255,7 +255,7 @@ public class MQTTTest {
         javax.jms.Topic jmsTopic = s.createTopic("foo.far");
         MessageProducer producer = s.createProducer(jmsTopic);
 
-        Topic[] topics = {new Topic(utf8("foo/far"), QoS.AT_MOST_ONCE)};
+        Topic[] topics = {new Topic(utf8("foo/+"), QoS.AT_MOST_ONCE)};
         connection.subscribe(topics);
         for (int i = 0; i < numberOfMessages; i++) {
             String payload = "This is Test Message: " + i;