QPID-3132: Prevent threshold events triggering directly further threshold events

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

4 files changed, 61 insertions, 3 deletions

diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index 122c5b9c1a..812d856416 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -51,7 +51,7 @@ Message::Message(const framing::SequenceNumber& id) :
     frames(id), persistenceId(0), redelivered(false), loaded(false),
     staged(false), forcePersistentPolicy(false), publisher(0), adapter(0), 
     expiration(FAR_FUTURE), dequeueCallback(0),
-    inCallback(false), requiredCredit(0)
+    inCallback(false), requiredCredit(0), isManagementMessage(false)
 {}
 
 Message::Message(const Message& original) :
@@ -443,4 +443,7 @@ framing::FieldTable& Message::getOrInsertHeaders()
     return getProperties<MessageProperties>()->getApplicationHeaders();
 }
 
+bool Message::getIsManagementMessage() const { return isManagementMessage; }
+void Message::setIsManagementMessage(bool b) { isManagementMessage = b; }
+
 }} // namespace qpid::broker
diff --git a/cpp/src/qpid/broker/Message.h b/cpp/src/qpid/broker/Message.h
index 2d0de27823..0e9ae7ba12 100644
--- a/cpp/src/qpid/broker/Message.h
+++ b/cpp/src/qpid/broker/Message.h
@@ -159,7 +159,8 @@ public:
     void resetDequeueCompleteCallback();
 
     uint8_t getPriority() const;
-
+    bool getIsManagementMessage() const;
+    void setIsManagementMessage(bool b);
   private:
     MessageAdapter& getAdapter() const;
     void allDequeuesComplete();
@@ -186,6 +187,7 @@ public:
     bool inCallback;
 
     uint32_t requiredCredit;
+    bool isManagementMessage;
 };
 
 }}
diff --git a/cpp/src/qpid/broker/ThresholdAlerts.cpp b/cpp/src/qpid/broker/ThresholdAlerts.cpp
index 4f35884af8..decb98df39 100644
--- a/cpp/src/qpid/broker/ThresholdAlerts.cpp
+++ b/cpp/src/qpid/broker/ThresholdAlerts.cpp
@@ -28,6 +28,52 @@
 
 namespace qpid {
 namespace broker {
+namespace {
+const qmf::org::apache::qpid::broker::EventQueueThresholdExceeded EVENT("dummy", 0, 0);
+bool isQMFv2(const boost::intrusive_ptr<Message> message)
+{
+    const qpid::framing::MessageProperties* props = message->getProperties<qpid::framing::MessageProperties>();
+    return props && props->getAppId() == "qmf2";
+}
+
+bool isThresholdEvent(const boost::intrusive_ptr<Message> message)
+{
+    if (message->getIsManagementMessage()) {
+        //is this a qmf event? if so is it a threshold event?
+        if (isQMFv2(message)) {
+            const qpid::framing::FieldTable* headers = message->getApplicationHeaders();
+            if (headers && headers->getAsString("qmf.content") == "_event") {
+                //decode as list
+                std::string content = message->getFrames().getContent();
+                qpid::types::Variant::List list;
+                qpid::amqp_0_10::ListCodec::decode(content, list);
+                if (list.empty() || list.front().getType() != qpid::types::VAR_MAP) return false;
+                qpid::types::Variant::Map map = list.front().asMap();
+                try {
+                    std::string eventName = map["_schema_id"].asMap()["_class_name"].asString();
+                    return eventName == EVENT.getEventName();
+                } catch (const std::exception& e) {
+                    QPID_LOG(error, "Error checking for recursive threshold alert: " << e.what());
+                }
+            }
+        } else {
+            std::string content = message->getFrames().getContent();
+            qpid::framing::Buffer buffer(const_cast<char*>(content.data()), content.size());
+            if (buffer.getOctet() == 'A' && buffer.getOctet() == 'M' && buffer.getOctet() == '2' && buffer.getOctet() == 'e') {
+                buffer.getLong();//sequence
+                std::string packageName;
+                buffer.getShortString(packageName);
+                if (packageName != EVENT.getPackageName()) return false;
+                std::string eventName;
+                buffer.getShortString(eventName);
+                return eventName == EVENT.getEventName();
+            }
+        }
+    }
+    return false;
+}
+}
+
 ThresholdAlerts::ThresholdAlerts(const std::string& n,
                                  qpid::management::ManagementAgent& a,
                                  const uint32_t ct,
@@ -44,8 +90,13 @@ void ThresholdAlerts::enqueued(const QueuedMessage& m)
     if ((countThreshold && count >= countThreshold) || (sizeThreshold && size >= sizeThreshold)) {
         if ((repeatInterval == 0 && lastAlert == qpid::sys::EPOCH)
             || qpid::sys::Duration(lastAlert, qpid::sys::now()) > repeatInterval) {
-            agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdExceeded(name, count, size));
+            //Note: Raising an event may result in messages being
+            //enqueued on queues; it may even be that this event
+            //causes a message to be enqueued on the queue we are
+            //tracking, and so we need to avoid recursing
+            if (isThresholdEvent(m.payload)) return;
             lastAlert = qpid::sys::now();
+            agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdExceeded(name, count, size));
         }
     }
 }
diff --git a/cpp/src/qpid/management/ManagementAgent.cpp b/cpp/src/qpid/management/ManagementAgent.cpp
index f88b9c0931..7c9771fe79 100644
--- a/cpp/src/qpid/management/ManagementAgent.cpp
+++ b/cpp/src/qpid/management/ManagementAgent.cpp
@@ -548,6 +548,7 @@ void ManagementAgent::sendBufferLH(Buffer&  buf,
     dp->setRoutingKey(routingKey);
 
     msg->getFrames().append(content);
+    msg->setIsManagementMessage(true);
 
     {
         sys::Mutex::ScopedUnlock u(userLock);
@@ -624,6 +625,7 @@ void ManagementAgent::sendBufferLH(const string& data,
         msg->setTimestamp(broker->getExpiryPolicy());
     }
     msg->getFrames().append(content);
+    msg->setIsManagementMessage(true);
 
     {
         sys::Mutex::ScopedUnlock u(userLock);