QPID-5729: handle concurrent triggering of topic policy

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

3 files changed, 16 insertions, 1 deletions

diff --git a/qpid/cpp/src/qpid/broker/amqp/NodePolicy.cpp b/qpid/cpp/src/qpid/broker/amqp/NodePolicy.cpp
index 6cefe36f67..6546102ca7 100644
--- a/qpid/cpp/src/qpid/broker/amqp/NodePolicy.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/NodePolicy.cpp
@@ -195,7 +195,7 @@ std::pair<boost::shared_ptr<Queue>, boost::shared_ptr<Topic> > TopicPolicy::crea
     qpid::amqp_0_10::translate(exchangeSettings, args);
     boost::shared_ptr<Exchange> exchange = connection.getBroker().createExchange(name, exchangeType, isDurable(), autodelete, alternateExchange,
                                                                                  args, connection.getUserId(), connection.getId()).first;
-    result.second = connection.getTopics().createTopic(connection.getBroker(), name, exchange, topicSettings);
+    result.second = connection.getTopics().declare(connection.getBroker(), name, exchange, topicSettings);
     return result;
 }
 
diff --git a/qpid/cpp/src/qpid/broker/amqp/Topic.cpp b/qpid/cpp/src/qpid/broker/amqp/Topic.cpp
index 7786180765..ec0fc617b6 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Topic.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Topic.cpp
@@ -115,6 +115,20 @@ boost::shared_ptr<Topic> TopicRegistry::createTopic(Broker& broker, const std::s
     return topic;
 }
 
+boost::shared_ptr<Topic> TopicRegistry::declare(Broker& broker, const std::string& name, boost::shared_ptr<Exchange> exchange, const qpid::types::Variant::Map& properties)
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+    Topics::const_iterator i = topics.find(name);
+    if (i == topics.end()) {
+        boost::shared_ptr<Topic> topic(new Topic(broker, name, exchange, properties));
+        topics.insert(Topics::value_type(name, topic));
+        topic->getExchange()->setDeletionListener(name, boost::bind(&TopicRegistry::remove, this, name));
+        return topic;
+    } else {
+        return i->second;
+    }
+}
+
 bool TopicRegistry::createObject(Broker& broker, const std::string& type, const std::string& name, const qpid::types::Variant::Map& props,
                                  const std::string& /*userId*/, const std::string& /*connectionId*/)
 {
diff --git a/qpid/cpp/src/qpid/broker/amqp/Topic.h b/qpid/cpp/src/qpid/broker/amqp/Topic.h
index df16f4a738..457b8cfced 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Topic.h
+++ b/qpid/cpp/src/qpid/broker/amqp/Topic.h
@@ -78,6 +78,7 @@ class TopicRegistry : public ObjectFactory
     boost::shared_ptr<Topic> remove(const std::string& name);
     boost::shared_ptr<Topic> get(const std::string& name);
     boost::shared_ptr<Topic> createTopic(Broker&, const std::string& name, boost::shared_ptr<Exchange> exchange, const qpid::types::Variant::Map& properties);
+    boost::shared_ptr<Topic> declare(Broker&, const std::string& name, boost::shared_ptr<Exchange> exchange, const qpid::types::Variant::Map& properties);
   private:
     typedef std::map<std::string, boost::shared_ptr<Topic> > Topics;
     qpid::sys::Mutex lock;