qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kgiu...@apache.org
Subject svn commit: r1179248 - in /qpid/branches/qpid-3346/qpid: cpp/src/qpid/broker/Broker.cpp cpp/src/qpid/broker/Broker.h cpp/src/qpid/broker/MessageGroupManager.cpp cpp/src/qpid/broker/MessageGroupManager.h tests/src/py/qpid_tests/broker_0_10/msg_groups.py
Date Wed, 05 Oct 2011 14:52:12 GMT
Author: kgiusti
Date: Wed Oct  5 14:52:11 2011
New Revision: 1179248

URL: http://svn.apache.org/viewvc?rev=1179248&view=rev
Log:
QPID-3346: allow configuration of the default group identifier

Modified:
    qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp
    qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h
    qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp
    qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.h
    qpid/branches/qpid-3346/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py

Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp?rev=1179248&r1=1179247&r2=1179248&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.cpp Wed Oct  5 14:52:11 2011
@@ -33,6 +33,7 @@
 #include "qpid/broker/Link.h"
 #include "qpid/broker/ExpiryPolicy.h"
 #include "qpid/broker/QueueFlowLimit.h"
+#include "qpid/broker/MessageGroupManager.h"
 
 #include "qmf/org/apache/qpid/broker/Package.h"
 #include "qmf/org/apache/qpid/broker/ArgsBrokerCreate.h"
@@ -123,7 +124,8 @@ Broker::Options::Options(const std::stri
     qmf1Support(true),
     queueFlowStopRatio(80),
     queueFlowResumeRatio(70),
-    queueThresholdEventRatio(80)
+    queueThresholdEventRatio(80),
+    defaultMsgGroup("qpid.no-group")
 {
     int c = sys::SystemInfo::concurrency();
     workerThreads=c+1;
@@ -159,7 +161,8 @@ Broker::Options::Options(const std::stri
         ("async-queue-events", optValue(asyncQueueEvents, "yes|no"), "Set Queue Events async,
used for services like replication")
         ("default-flow-stop-threshold", optValue(queueFlowStopRatio, "PERCENT"), "Percent
of queue's maximum capacity at which flow control is activated.")
         ("default-flow-resume-threshold", optValue(queueFlowResumeRatio, "PERCENT"), "Percent
of queue's maximum capacity at which flow control is de-activated.")
-        ("default-event-threshold-ratio", optValue(queueThresholdEventRatio, "%age of limit"),
"The ratio of any specified queue limit at which an event will be raised");
+        ("default-event-threshold-ratio", optValue(queueThresholdEventRatio, "%age of limit"),
"The ratio of any specified queue limit at which an event will be raised")
+        ("default-message-group", optValue(defaultMsgGroup, "GROUP-IDENTIFER"), "Group identifier
to assign to messages delivered to a message group queue that do not contain an identifier.");
 }
 
 const std::string empty;
@@ -250,6 +253,7 @@ Broker::Broker(const Broker::Options& co
     Plugin::earlyInitAll(*this);
 
     QueueFlowLimit::setDefaults(conf.queueLimit, conf.queueFlowStopRatio, conf.queueFlowResumeRatio);
+    MessageGroupManager::setDefaults(conf.defaultMsgGroup);
 
     // If no plugin store module registered itself, set up the null store.
     if (NullMessageStore::isNullStore(store.get()))

Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h?rev=1179248&r1=1179247&r2=1179248&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/Broker.h Wed Oct  5 14:52:11 2011
@@ -121,6 +121,7 @@ public:
         uint queueFlowStopRatio;    // producer flow control: on
         uint queueFlowResumeRatio;  // producer flow control: off
         uint16_t queueThresholdEventRatio;
+        std::string defaultMsgGroup;
 
       private:
         std::string getHome();

Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp?rev=1179248&r1=1179247&r2=1179248&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.cpp Wed Oct  5 14:52:11
2011
@@ -41,15 +41,14 @@ namespace {
 const std::string MessageGroupManager::qpidMessageGroupKey("qpid.group_header_key");
 const std::string MessageGroupManager::qpidSharedGroup("qpid.shared_msg_group");
 const std::string MessageGroupManager::qpidMessageGroupTimestamp("qpid.group_timestamp");
-const std::string MessageGroupManager::qpidMessageGroupDefault("qpid.no_group");     /**
@todo KAG: make configurable in Broker options */
 
 
 const std::string MessageGroupManager::getGroupId( const QueuedMessage& qm ) const
 {
     const qpid::framing::FieldTable* headers = qm.payload->getApplicationHeaders();
-    if (!headers) return qpidMessageGroupDefault;
+    if (!headers) return defaultGroupId;
     qpid::framing::FieldTable::ValuePtr id = headers->get( groupIdHeader );
-    if (!id || !id->convertsTo<std::string>()) return qpidMessageGroupDefault;
+    if (!id || !id->convertsTo<std::string>()) return defaultGroupId;
     return id->get<std::string>();
 }
 
@@ -331,6 +330,12 @@ boost::shared_ptr<MessageGroupManager> M
     return empty;
 }
 
+std::string MessageGroupManager::defaultGroupId;
+void MessageGroupManager::setDefaults(const std::string& groupId)   // static
+{
+    defaultGroupId = groupId;
+}
+
 /** Cluster replication:
 
    state map format:

Modified: qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.h
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.h?rev=1179248&r1=1179247&r2=1179248&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.h (original)
+++ qpid/branches/qpid-3346/qpid/cpp/src/qpid/broker/MessageGroupManager.h Wed Oct  5 14:52:11
2011
@@ -36,6 +36,8 @@ class MessageDistributor;
 
 class MessageGroupManager : public StatefulQueueObserver, public MessageDistributor
 {
+    static std::string defaultGroupId;  // assigned if no group id header present
+
     const std::string groupIdHeader;    // msg header holding group identifier
     const unsigned int timestamp;       // mark messages with timestamp if set
     Messages& messages;                 // parent Queue's in memory message container
@@ -65,7 +67,6 @@ class MessageGroupManager : public State
     static const std::string qpidMessageGroupKey;
     static const std::string qpidSharedGroup;   // if specified, one group can be consumed
by multiple receivers
     static const std::string qpidMessageGroupTimestamp;
-    static const std::string qpidMessageGroupDefault;
 
     const std::string getGroupId( const QueuedMessage& qm ) const;
     void unFree( const GroupState& state )
@@ -92,6 +93,7 @@ class MessageGroupManager : public State
 
  public:
 
+    static QPID_BROKER_EXTERN void setDefaults(const std::string& groupId);
     static boost::shared_ptr<MessageGroupManager> create( const std::string& qName,
                                                           Messages& messages,
                                                           const qpid::framing::FieldTable&
settings );

Modified: qpid/branches/qpid-3346/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3346/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py?rev=1179248&r1=1179247&r2=1179248&view=diff
==============================================================================
--- qpid/branches/qpid-3346/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py (original)
+++ qpid/branches/qpid-3346/qpid/tests/src/py/qpid_tests/broker_0_10/msg_groups.py Wed Oct
 5 14:52:11 2011
@@ -936,6 +936,42 @@ class MultiConsumerMsgGroupTests(Base):
         snd.close()
         self.ssn.close()
 
+    def test_default_group_id(self):
+        """ Verify the queue assigns the default group id should a message
+        arrive without a group identifier.
+        """
+        snd = self.ssn.sender("msg-group-q; {create:always, delete:sender," +
+                              " node: {x-declare: {arguments:" +
+                              " {'qpid.group_header_key':'THE-GROUP'," +
+                              "'qpid.shared_msg_group':1}}}}")
+
+        m = Message(content={}, properties={"NO-GROUP-HEADER":"HA-HA"})
+        snd.send(m)
+
+        # now setup a QMF session, so we can call methods
+        self.qmf_session = qmf.console.Session()
+        self.qmf_broker = self.qmf_session.addBroker(str(self.broker))
+        brokers = self.qmf_session.getObjects(_class="broker")
+        assert len(brokers) == 1
+        broker = brokers[0]
+
+        # grab the group state off the queue, and verify the default group is
+        # present ("qpid.no-group" is the broker default)
+        rc = broker.query("queue", "msg-group-q")
+        assert rc.status == 0
+        assert rc.text == "OK"
+        results = rc.outArgs['results']
+        assert 'qpid.message_group_queue' in results
+        q_info = results['qpid.message_group_queue']
+        assert 'group_header_key' in q_info and q_info['group_header_key'] == "THE-GROUP"
+        assert 'group_state' in q_info and len(q_info['group_state']) == 1
+        g_info = q_info['group_state'][0]
+        assert 'group_id' in g_info
+        assert g_info['group_id'] == 'qpid.no-group'
+
+        self.qmf_session.delBroker(self.qmf_broker)
+
+
 class StickyConsumerMsgGroupTests(Base):
     """
     Tests for the behavior of sticky-consumer message groups.  These tests



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message