From dad7a9e346d47a28dfe6f69fe95f9ee7129d0d22 Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Wed, 9 Mar 2011 19:24:05 +0000 Subject: QPID-3132: merged to 0.10 branch. git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/0.10@1079953 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/qpid/broker/Message.cpp | 5 +- qpid/cpp/src/qpid/broker/Message.h | 4 +- qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp | 53 +++++++++++++++++++++- qpid/cpp/src/qpid/management/ManagementAgent.cpp | 2 + .../src/py/qpid_tests/broker_0_10/threshold.py | 15 ++++++ 5 files changed, 76 insertions(+), 3 deletions(-) diff --git a/qpid/cpp/src/qpid/broker/Message.cpp b/qpid/cpp/src/qpid/broker/Message.cpp index 122c5b9c1a..812d856416 100644 --- a/qpid/cpp/src/qpid/broker/Message.cpp +++ b/qpid/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()->getApplicationHeaders(); } +bool Message::getIsManagementMessage() const { return isManagementMessage; } +void Message::setIsManagementMessage(bool b) { isManagementMessage = b; } + }} // namespace qpid::broker diff --git a/qpid/cpp/src/qpid/broker/Message.h b/qpid/cpp/src/qpid/broker/Message.h index 2d0de27823..0e9ae7ba12 100644 --- a/qpid/cpp/src/qpid/broker/Message.h +++ b/qpid/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/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp b/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp index 4f35884af8..decb98df39 100644 --- a/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp +++ b/qpid/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) +{ + const qpid::framing::MessageProperties* props = message->getProperties(); + return props && props->getAppId() == "qmf2"; +} + +bool isThresholdEvent(const boost::intrusive_ptr 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(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/qpid/cpp/src/qpid/management/ManagementAgent.cpp b/qpid/cpp/src/qpid/management/ManagementAgent.cpp index f88b9c0931..7c9771fe79 100644 --- a/qpid/cpp/src/qpid/management/ManagementAgent.cpp +++ b/qpid/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); diff --git a/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py b/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py index bcd3c507e2..6628ae8424 100644 --- a/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py +++ b/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py @@ -60,3 +60,18 @@ class ThresholdTests (Base): def test_alert_size_alias(self): self.do_threshold_test("x-qpid-maximum-message-size", 15, [Message("msg-%s" % i) for i in range(3)]) + + def test_alert_on_alert_queue(self): + rcv = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdExceeded.#; {link:{x-declare:{arguments:{'qpid.alert_count':1}}}}") + rcvQMFv1 = self.ssn.receiver("qpid.management/console.event.#; {link:{x-declare:{arguments:{'qpid.alert_count':1}}}}") + snd = self.ssn.sender("ttq; {create:always, node: {x-declare:{auto_delete:True,exclusive:True,arguments:{'qpid.alert_count':1}}}}") + snd.send(Message("my-message")) + queues = [] + for i in range(2): + event = rcv.fetch() + schema = event.content[0]["_schema_id"] + assert schema["_class_name"] == "queueThresholdExceeded" + values = event.content[0]["_values"] + queues.append(values["qName"]) + assert "ttq" in queues, "expected event for ttq (%s)" % (queues) + -- cgit v1.2.1