Author: kgiusti
Date: Wed Mar 9 19:24:05 2011
New Revision: 1079953
URL: http://svn.apache.org/viewvc?rev=1079953&view=rev
Log:
QPID-3132: merged to 0.10 branch.
Modified:
qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.cpp
qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.h
qpid/branches/0.10/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp
qpid/branches/0.10/qpid/cpp/src/qpid/management/ManagementAgent.cpp
(contents, props changed)
qpid/branches/0.10/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py
Modified: qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.cpp
URL:
http://svn.apache.org/viewvc/qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.cpp?rev=1079953&r1=1079952&r2=1079953&view=diff
==============================================================================
--- qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.cpp (original)
+++ qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.cpp Wed Mar 9 19:24:05
2011
@@ -51,7 +51,7 @@ Message::Message(const framing::Sequence
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::getOrInser
return getProperties<MessageProperties>()->getApplicationHeaders();
}
+bool Message::getIsManagementMessage() const { return isManagementMessage; }
+void Message::setIsManagementMessage(bool b) { isManagementMessage = b; }
+
}} // namespace qpid::broker
Modified: qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.h
URL:
http://svn.apache.org/viewvc/qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.h?rev=1079953&r1=1079952&r2=1079953&view=diff
==============================================================================
--- qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.h (original)
+++ qpid/branches/0.10/qpid/cpp/src/qpid/broker/Message.h Wed Mar 9 19:24:05
2011
@@ -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;
};
}}
Modified: qpid/branches/0.10/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp
URL:
http://svn.apache.org/viewvc/qpid/branches/0.10/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp?rev=1079953&r1=1079952&r2=1079953&view=diff
==============================================================================
--- qpid/branches/0.10/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp (original)
+++ qpid/branches/0.10/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp Wed Mar 9
19:24:05 2011
@@ -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 Que
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));
}
}
}
Modified: qpid/branches/0.10/qpid/cpp/src/qpid/management/ManagementAgent.cpp
URL:
http://svn.apache.org/viewvc/qpid/branches/0.10/qpid/cpp/src/qpid/management/ManagementAgent.cpp?rev=1079953&r1=1079952&r2=1079953&view=diff
==============================================================================
--- qpid/branches/0.10/qpid/cpp/src/qpid/management/ManagementAgent.cpp
(original)
+++ qpid/branches/0.10/qpid/cpp/src/qpid/management/ManagementAgent.cpp Wed Mar
9 19:24:05 2011
@@ -548,6 +548,7 @@ void ManagementAgent::sendBufferLH(Buffe
dp->setRoutingKey(routingKey);
msg->getFrames().append(content);
+ msg->setIsManagementMessage(true);
{
sys::Mutex::ScopedUnlock u(userLock);
@@ -624,6 +625,7 @@ void ManagementAgent::sendBufferLH(const
msg->setTimestamp(broker->getExpiryPolicy());
}
msg->getFrames().append(content);
+ msg->setIsManagementMessage(true);
{
sys::Mutex::ScopedUnlock u(userLock);
Propchange: qpid/branches/0.10/qpid/cpp/src/qpid/management/ManagementAgent.cpp
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Wed Mar 9 19:24:05 2011
@@ -1,2 +1,2 @@
/qpid/branches/qpid-2935/qpid/cpp/src/qpid/management/ManagementAgent.cpp:1061302-1072333
-/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp:1078075,1078733,1078743,1079385,1079590,1079830
+/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp:1078075,1078733,1078743,1079385,1079590,1079830,1079854
Modified:
qpid/branches/0.10/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py
URL:
http://svn.apache.org/viewvc/qpid/branches/0.10/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py?rev=1079953&r1=1079952&r2=1079953&view=diff
==============================================================================
--- qpid/branches/0.10/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py
(original)
+++ qpid/branches/0.10/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py
Wed Mar 9 19:24:05 2011
@@ -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)
+
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
