Return-Path: X-Original-To: apmail-qpid-commits-archive@www.apache.org Delivered-To: apmail-qpid-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id 3C4DCEE77 for ; Fri, 8 Mar 2013 22:06:52 +0000 (UTC) Received: (qmail 27480 invoked by uid 500); 8 Mar 2013 22:06:52 -0000 Delivered-To: apmail-qpid-commits-archive@qpid.apache.org Received: (qmail 27439 invoked by uid 500); 8 Mar 2013 22:06:52 -0000 Mailing-List: contact commits-help@qpid.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@qpid.apache.org Delivered-To: mailing list commits@qpid.apache.org Received: (qmail 27428 invoked by uid 99); 8 Mar 2013 22:06:52 -0000 Received: from athena.apache.org (HELO athena.apache.org) (140.211.11.136) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 08 Mar 2013 22:06:52 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=5.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [140.211.11.4] (HELO eris.apache.org) (140.211.11.4) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 08 Mar 2013 22:06:50 +0000 Received: from eris.apache.org (localhost [127.0.0.1]) by eris.apache.org (Postfix) with ESMTP id EE890238897F; Fri, 8 Mar 2013 22:06:30 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r1454601 - in /qpid/trunk/qpid: cpp/src/qpid/broker/ specs/ tests/src/py/qpid_tests/broker_0_10/ Date: Fri, 08 Mar 2013 22:06:30 -0000 To: commits@qpid.apache.org From: tross@apache.org X-Mailer: svnmailer-1.0.8-patched Message-Id: <20130308220630.EE890238897F@eris.apache.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: tross Date: Fri Mar 8 22:06:30 2013 New Revision: 1454601 URL: http://svn.apache.org/r1454601 Log: QPID-4632 - Improvement to queue threshold alerting Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.h qpid/trunk/qpid/specs/management-schema.xml qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp?rev=1454601&r1=1454600&r2=1454601&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp (original) +++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp Fri Mar 8 22:06:30 2013 @@ -47,6 +47,10 @@ const std::string AUTO_DELETE_TIMEOUT("q const std::string ALERT_REPEAT_GAP("qpid.alert_repeat_gap"); const std::string ALERT_COUNT("qpid.alert_count"); const std::string ALERT_SIZE("qpid.alert_size"); +const std::string ALERT_COUNT_UP("qpid.alert_count_up"); +const std::string ALERT_SIZE_UP("qpid.alert_size_up"); +const std::string ALERT_COUNT_DOWN("qpid.alert_count_down"); +const std::string ALERT_SIZE_DOWN("qpid.alert_size_down"); const std::string PRIORITIES("qpid.priorities"); const std::string FAIRSHARE("qpid.fairshare"); const std::string FAIRSHARE_ALIAS("x-qpid-fairshare"); @@ -165,12 +169,18 @@ bool QueueSettings::handle(const std::st } else if (key == ALERT_REPEAT_GAP) { alertRepeatInterval = value; return true; - } else if (key == ALERT_COUNT) { + } else if ((key == ALERT_COUNT) || (key == ALERT_COUNT_UP)) { alertThreshold.setCount(value); return true; - } else if (key == ALERT_SIZE) { + } else if ((key == ALERT_SIZE) || (key == ALERT_SIZE_UP)) { alertThreshold.setSize(value); return true; + } else if (key == ALERT_COUNT_DOWN) { + alertThresholdDown.setCount(value); + return true; + } else if (key == ALERT_SIZE_DOWN) { + alertThresholdDown.setSize(value); + return true; } else if (key == MAX_FILE_COUNT && value.asUint64() > 0) { maxFileCount = value.asUint64(); return false; // 'handle' here and also pass to store Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h?rev=1454601&r1=1454600&r2=1454601&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h (original) +++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h Fri Mar 8 22:06:30 2013 @@ -71,6 +71,7 @@ struct QueueSettings //threshold events: QueueDepth alertThreshold; + QueueDepth alertThresholdDown; int64_t alertRepeatInterval; //file limits checked by Acl and shared with storeSettings Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp?rev=1454601&r1=1454600&r2=1454601&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp (original) +++ qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.cpp Fri Mar 8 22:06:30 2013 @@ -21,84 +21,44 @@ #include "qpid/broker/ThresholdAlerts.h" #include "qpid/broker/Queue.h" #include "qpid/broker/Message.h" -#include "qpid/broker/amqp_0_10/MessageTransfer.h" -#include "qpid/amqp_0_10/Codecs.h" #include "qpid/log/Statement.h" #include "qpid/management/ManagementAgent.h" +#include "qmf/org/apache/qpid/broker/EventQueueThresholdCrossedUpward.h" +#include "qmf/org/apache/qpid/broker/EventQueueThresholdCrossedDownward.h" #include "qmf/org/apache/qpid/broker/EventQueueThresholdExceeded.h" namespace qpid { namespace broker { -namespace { -const qmf::org::apache::qpid::broker::EventQueueThresholdExceeded EVENT("dummy", 0, 0); -bool isQMFv2(const Message& message) -{ - const qpid::framing::MessageProperties* props = qpid::broker::amqp_0_10::MessageTransfer::get(message).getProperties(); - return props && props->getAppId() == "qmf2"; -} - -bool isThresholdEvent(const Message& message) -{ - if (message.getIsManagementMessage()) { - //is this a qmf event? if so is it a threshold event? - if (isQMFv2(message)) { - if (message.getPropertyAsString("qmf.content") == "_event") { - //decode as list - std::string content = qpid::broker::amqp_0_10::MessageTransfer::get(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 = qpid::broker::amqp_0_10::MessageTransfer::get(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, - const uint64_t st, - const long repeat) - : name(n), agent(a), countThreshold(ct), sizeThreshold(st), - repeatInterval(repeat ? repeat*qpid::sys::TIME_SEC : 0), - count(0), size(0), lastAlert(qpid::sys::EPOCH) {} + const uint32_t ctu, + const uint32_t ctd, + const uint64_t stu, + const uint64_t std, + const bool bw) + : name(n), agent(a), + countThreshold(ctu), countThresholdDown(ctd), + sizeThreshold(stu), sizeThresholdDown(std), + count(0), size(0), countGoingUp(true), sizeGoingUp(true), backwardCompat(bw) {} void ThresholdAlerts::enqueued(const Message& m) { size += m.getContentSize(); ++count; - if ((countThreshold && count >= countThreshold) || (sizeThreshold && size >= sizeThreshold)) { - if ((repeatInterval == 0 && lastAlert == qpid::sys::EPOCH) - || qpid::sys::Duration(lastAlert, qpid::sys::now()) > repeatInterval) { - //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)) return; - lastAlert = qpid::sys::now(); + + if (sizeGoingUp && sizeThreshold && size >= sizeThreshold) { + sizeGoingUp = false; + agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdCrossedUpward(name, count, size)); + if (backwardCompat) + agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdExceeded(name, count, size)); + } + + if (countGoingUp && countThreshold && count >= countThreshold) { + countGoingUp = false; + agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdCrossedUpward(name, count, size)); + if (backwardCompat) agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdExceeded(name, count, size)); - QPID_LOG(info, "Threshold event triggered for " << name << ", count=" << count << ", size=" << size); - } } } @@ -106,21 +66,32 @@ void ThresholdAlerts::dequeued(const Mes { size -= m.getContentSize(); --count; - if ((countThreshold && count < countThreshold) || (sizeThreshold && size < sizeThreshold)) { - lastAlert = qpid::sys::EPOCH; + + if (!sizeGoingUp && sizeThreshold && size <= sizeThresholdDown) { + sizeGoingUp = true; + agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdCrossedDownward(name, count, size)); + } + + if (!countGoingUp && countThreshold && count <= countThresholdDown) { + countGoingUp = true; + agent.raiseEvent(qmf::org::apache::qpid::broker::EventQueueThresholdCrossedDownward(name, count, size)); } } void ThresholdAlerts::observe(Queue& queue, qpid::management::ManagementAgent& agent, - const uint64_t countThreshold, - const uint64_t sizeThreshold, - const long repeatInterval) -{ - if (countThreshold || sizeThreshold) { + const uint64_t ctu, + const uint64_t _ctd, + const uint64_t stu, + const uint64_t _std) +{ + if (ctu || stu) { + uint64_t ctd = (_ctd == 0 || _ctd >= ctu) ? ctu >> 1 : _ctd; + uint64_t std = (_std == 0 || _std >= stu) ? stu >> 1 : _std; + boost::shared_ptr observer( - new ThresholdAlerts(queue.getName(), agent, countThreshold, sizeThreshold, repeatInterval) + new ThresholdAlerts(queue.getName(), agent, ctu, ctd, stu, std, (_ctd == 0 && _std == 0)) ); queue.addObserver(observer); } @@ -133,8 +104,10 @@ void ThresholdAlerts::observe(Queue& que //percentage of any limit from the policy. uint32_t countThreshold = settings.alertThreshold.hasCount() ? settings.alertThreshold.getCount() : (settings.maxDepth.getCount()*limitRatio/100); uint32_t sizeThreshold = settings.alertThreshold.hasSize() ? settings.alertThreshold.getSize() : (settings.maxDepth.getSize()*limitRatio/100); + uint32_t countThresholdDown = settings.alertThresholdDown.hasCount() ? settings.alertThresholdDown.getCount() : 0; + uint32_t sizeThresholdDown = settings.alertThresholdDown.hasSize() ? settings.alertThresholdDown.getSize() : 0; - observe(queue, agent, countThreshold, sizeThreshold, settings.alertRepeatInterval); + observe(queue, agent, countThreshold, countThresholdDown , sizeThreshold, sizeThresholdDown); } }} Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.h URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.h?rev=1454601&r1=1454600&r2=1454601&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.h (original) +++ qpid/trunk/qpid/cpp/src/qpid/broker/ThresholdAlerts.h Fri Mar 8 22:06:30 2013 @@ -22,7 +22,6 @@ * */ #include "qpid/broker/QueueObserver.h" -#include "qpid/sys/Time.h" #include "qpid/types/Variant.h" #include @@ -44,8 +43,10 @@ class ThresholdAlerts : public QueueObse ThresholdAlerts(const std::string& name, qpid::management::ManagementAgent& agent, const uint32_t countThreshold, + const uint32_t countThresholdDown, const uint64_t sizeThreshold, - const long repeatInterval); + const uint64_t sizeThresholdDown, + const bool backwardCompat); void enqueued(const Message&); void dequeued(const Message&); void acquired(const Message&) {}; @@ -53,19 +54,23 @@ class ThresholdAlerts : public QueueObse static void observe(Queue& queue, qpid::management::ManagementAgent& agent, const uint64_t countThreshold, + const uint64_t countThresholdDown, const uint64_t sizeThreshold, - const long repeatInterval); + const uint64_t sizeThresholdDown); static void observe(Queue& queue, qpid::management::ManagementAgent& agent, const QueueSettings& settings, uint16_t limitRatio); private: const std::string name; qpid::management::ManagementAgent& agent; const uint32_t countThreshold; + const uint32_t countThresholdDown; const uint64_t sizeThreshold; - const qpid::sys::Duration repeatInterval; + const uint64_t sizeThresholdDown; uint64_t count; uint64_t size; - qpid::sys::AbsTime lastAlert; + bool countGoingUp; + bool sizeGoingUp; + bool backwardCompat; }; }} // namespace qpid::broker Modified: qpid/trunk/qpid/specs/management-schema.xml URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/specs/management-schema.xml?rev=1454601&r1=1454600&r2=1454601&view=diff ============================================================================== --- qpid/trunk/qpid/specs/management-schema.xml (original) +++ qpid/trunk/qpid/specs/management-schema.xml Fri Mar 8 22:06:30 2013 @@ -523,6 +523,10 @@ - + + + + + Modified: qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py?rev=1454601&r1=1454600&r2=1454601&view=diff ============================================================================== --- qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py (original) +++ qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/threshold.py Fri Mar 8 22:06:30 2013 @@ -32,46 +32,181 @@ class ThresholdTests (Base): def setup_session(self): return self.conn.session() - def do_threshold_test(self, key, value, messages): - rcv = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdExceeded.#") - snd = self.ssn.sender("ttq; {create:always, node: {x-declare:{auto_delete:True,exclusive:True,arguments:{'%s':%s}}}}" % (key, value)) + def enqueue(self, snd, count): + for i in range(count): + m = Message("msg-%d" % i) + snd.send(m) + + def dequeue(self, rcv, count): + for i in range(count): + m = rcv.fetch(timeout=1) + self.ssn.acknowledge() + + def check_events(self, rcv, count): + for i in range(count): + m = rcv.fetch(timeout=0) + try: + m = rcv.fetch(timeout=0) + assert False + except: + pass + + def do_threshold_test(self, args, messages, drain_count, bw_compat=None): + astr = '' + first = True + for key, value in args.items(): + if first: + first = None + else: + astr += ',' + astr += "'%s':%s" % (key, value) + rcvUp = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedUpward.#") + rcvDn = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedDownward.#") + rcvBw = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdExceeded.#") + snd = self.ssn.sender("ttq; {create:always, node: {x-declare:{auto_delete:True,exclusive:True,arguments:{%s}}}}" % astr) + rcv = self.ssn.receiver("ttq") size = 0 count = 0 for m in messages: snd.send(m) count = count + 1 size = size + len(m.content) - event = rcv.fetch(timeout=1) + event = rcvUp.fetch(timeout=1) schema = event.content[0]["_schema_id"] - assert schema["_class_name"] == "queueThresholdExceeded" + assert schema["_class_name"] == "queueThresholdCrossedUpward" values = event.content[0]["_values"] assert values["qName"] == "ttq" assert values["msgDepth"] == count, "msgDepth %s, expected %s" % (values["msgDepth"], count) assert values["byteDepth"] == size, "byteDepth %s, expected %s" % (values["byteDepth"], size) + if bw_compat: + event = rcvBw.fetch(timeout=0) + + try: + event = rcvUp.fetch(timeout=0) + assert False + except: + pass + + if drain_count > 0: + for i in range(drain_count): + m = rcv.fetch(timeout=1) + self.ssn.acknowledge() + count -= 1 + size -= len(m.content) + event = rcvDn.fetch(timeout=1) + schema = event.content[0]["_schema_id"] + assert schema["_class_name"] == "queueThresholdCrossedDownward" + values = event.content[0]["_values"] + assert values["qName"] == "ttq" + assert values["msgDepth"] == count, "msgDepth %s, expected %s" % (values["msgDepth"], count) + assert values["byteDepth"] == size, "byteDepth %s, expected %s" % (values["byteDepth"], size) + try: + event = rcvUp.fetch(timeout=0) + assert False + except: + pass def test_alert_count(self): - self.do_threshold_test("qpid.alert_count", 5, [Message("msg-%s" % i) for i in range(5)]) + a = {'qpid.alert_count':5, 'qpid.alert_count_down':3} + self.do_threshold_test(a, [Message("msg-%s" % i) for i in range(5)], 2) def test_alert_size(self): - self.do_threshold_test("qpid.alert_size", 25, [Message("msg-%s" % i) for i in range(5)]) + a = {'qpid.alert_size_up':25,'qpid.alert_size_down':15} + self.do_threshold_test(a, [Message("msg-%s" % i) for i in range(5)], 2) def test_alert_count_alias(self): - self.do_threshold_test("x-qpid-maximum-message-count", 10, [Message("msg-%s" % i) for i in range(10)]) + a = {'x-qpid-maximum-message-count':10} + self.do_threshold_test(a, [Message("msg-%s" % i) for i in range(10)], 0, True) 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)]) + a = {'x-qpid-maximum-message-size':15} + self.do_threshold_test(a, [Message("msg-%s" % i) for i in range(3)], 0, True) 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}}}}") + rcv = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedUpward.#; {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() + for i in range(3): + event = rcv.fetch(timeout=1) schema = event.content[0]["_schema_id"] - assert schema["_class_name"] == "queueThresholdExceeded" + assert schema["_class_name"] == "queueThresholdCrossedUpward" values = event.content[0]["_values"] queues.append(values["qName"]) assert "ttq" in queues, "expected event for ttq (%s)" % (queues) + def test_hysteresis(self): + astr = "'qpid.alert_count_up':10,'qpid.alert_count_down':5" + rcvUp = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedUpward.#") + rcvDn = self.ssn.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdCrossedDownward.#") + snd = self.ssn.sender("thq; {create:always, node: {x-declare:{auto_delete:True,exclusive:True,arguments:{%s}}}}" % astr) + rcv = self.ssn.receiver("thq") + + rcvUp.capacity = 5 + rcvDn.capacity = 5 + rcv.capacity = 5 + + self.enqueue(snd, 8) # depth = 8 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.dequeue(rcv, 6) # depth = 2 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.enqueue(snd, 8) # depth = 10 + self.check_events(rcvUp, 1) + self.check_events(rcvDn, 0) + + self.dequeue(rcv, 1) # depth = 9 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.enqueue(snd, 1) # depth = 10 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.enqueue(snd, 10) # depth = 20 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.dequeue(rcv, 5) # depth = 15 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.dequeue(rcv, 12) # depth = 3 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 1) + + self.dequeue(rcv, 1) # depth = 2 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.enqueue(snd, 6) # depth = 8 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.enqueue(snd, 6) # depth = 14 + self.check_events(rcvUp, 1) + self.check_events(rcvDn, 0) + + self.dequeue(rcv, 9) # depth = 5 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 1) + + self.enqueue(snd, 1) # depth = 6 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.dequeue(rcv, 1) # depth = 5 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + self.dequeue(rcv, 5) # depth = 0 + self.check_events(rcvUp, 0) + self.check_events(rcvDn, 0) + + + + --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org For additional commands, e-mail: commits-help@qpid.apache.org