You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2015/01/08 00:28:54 UTC
svn commit: r1650196 - in /qpid/trunk/qpid: cpp/src/CMakeLists.txt
cpp/src/qpid/broker/LossyQueue.h cpp/src/qpid/broker/Lvq.h
cpp/src/qpid/broker/QueueFactory.cpp
tests/src/py/qpid_tests/broker_0_10/lvq.py
Author: gsim
Date: Wed Jan 7 23:28:54 2015
New Revision: 1650196
URL: http://svn.apache.org/r1650196
Log:
QPID-6299: allow ring and lvq behaviours to be combined
Modified:
qpid/trunk/qpid/cpp/src/CMakeLists.txt
qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.h
qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.h
qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp
qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/lvq.py
Modified: qpid/trunk/qpid/cpp/src/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/CMakeLists.txt?rev=1650196&r1=1650195&r2=1650196&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/CMakeLists.txt (original)
+++ qpid/trunk/qpid/cpp/src/CMakeLists.txt Wed Jan 7 23:28:54 2015
@@ -1077,6 +1077,7 @@ set (qpidbroker_SOURCES
qpid/broker/IngressCompletion.cpp
qpid/broker/Link.cpp
qpid/broker/LinkRegistry.cpp
+ qpid/broker/LossyLvq.cpp
qpid/broker/LossyQueue.cpp
qpid/broker/Lvq.cpp
qpid/broker/Message.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.h?rev=1650196&r1=1650195&r2=1650196&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.h Wed Jan 7 23:28:54 2015
@@ -29,7 +29,7 @@ namespace broker {
/**
* Drops messages to prevent a breach of any configured maximum depth.
*/
-class LossyQueue : public Queue
+class LossyQueue : public virtual Queue
{
public:
LossyQueue(const std::string&, const QueueSettings&, MessageStore* const, management::Manageable*, Broker*);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.h?rev=1650196&r1=1650195&r2=1650196&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.h Wed Jan 7 23:28:54 2015
@@ -32,7 +32,7 @@ class MessageMap;
* conjunction with the MessageMap class. This requires an existing
* message to be 'replaced' by a newer message with the same key.
*/
-class Lvq : public Queue
+class Lvq : public virtual Queue
{
public:
Lvq(const std::string&, std::auto_ptr<MessageMap>, const QueueSettings&, MessageStore* const, management::Manageable*, Broker*);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp?rev=1650196&r1=1650195&r2=1650196&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp Wed Jan 7 23:28:54 2015
@@ -22,6 +22,7 @@
#include "qpid/broker/Broker.h"
#include "qpid/broker/QueueSettings.h"
#include "qpid/broker/Queue.h"
+#include "qpid/broker/LossyLvq.h"
#include "qpid/broker/LossyQueue.h"
#include "qpid/broker/Lvq.h"
#include "qpid/broker/Messages.h"
@@ -51,10 +52,17 @@ boost::shared_ptr<Queue> QueueFactory::c
boost::shared_ptr<QueueFlowLimit> flow_ptr(QueueFlowLimit::createLimit(name, settings));
//1. determine Queue type (i.e. whether we are subclassing Queue)
- // -> if 'ring' policy is in use then subclass
boost::shared_ptr<Queue> queue;
if (settings.dropMessagesAtLimit) {
- queue = boost::shared_ptr<Queue>(new LossyQueue(name, settings, settings.durable ? store : 0, parent, broker));
+ // -> if 'ring' policy is in use then subclass
+ if (settings.lvqKey.size()) {
+ //combination of ring and lvq:
+ std::auto_ptr<MessageMap> map(new MessageMap(settings.lvqKey));
+ queue = boost::shared_ptr<Queue>(new LossyLvq(name, map, settings, settings.durable ? store : 0, parent, broker));
+ } else {
+ //simple ring:
+ queue = boost::shared_ptr<Queue>(new LossyQueue(name, settings, settings.durable ? store : 0, parent, broker));
+ }
} else if (settings.selfDestructAtLimit) {
queue = boost::shared_ptr<Queue>(new SelfDestructQueue(name, settings, settings.durable ? store : 0, parent, broker));
} else if (settings.lvqKey.size()) {
Modified: qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/lvq.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/lvq.py?rev=1650196&r1=1650195&r2=1650196&view=diff
==============================================================================
--- qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/lvq.py (original)
+++ qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/lvq.py Wed Jan 7 23:28:54 2015
@@ -6,9 +6,9 @@
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
-#
+#
# http://www.apache.org/licenses/LICENSE-2.0
-#
+#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -20,18 +20,19 @@
from qpid.tests.messaging.implementation import *
from qpid.tests.messaging import Base
import math
+import random
class LVQTests (Base):
"""
Test last value queue behaviour
- """
+ """
def setup_connection(self):
return Connection.establish(self.broker, **self.connection_options())
def setup_session(self):
return self.conn.session()
-
+
def test_simple(self):
snd = self.ssn.sender("lvq; {create: sender, delete: sender, node: {x-declare:{arguments:{'qpid.last_value_queue_key':lvq-key}}}}",
durable=self.durable())
@@ -59,6 +60,42 @@ class LVQTests (Base):
rcv = self.ssn.receiver("lvq; {mode: browse}")
assert (fetch_all(rcv) == ["a-3", "c-3", "d-1", "b-3"])
+ def check_ring_lvq(self, ring_size, keys, message_count):
+ address = "lvq; {create: sender, delete: sender, node: {x-declare:{arguments:{'qpid.last_value_queue_key':lvq-key,'qpid.policy_type':'ring','qpid.max_count':%i}}}}" % ring_size
+ snd = self.ssn.sender(address, durable=self.durable())
+ counters = {}
+ for k in keys:
+ counters[k] = 0
+ messages = []
+ for i in range(message_count):
+ k = random.choice(keys)
+ counters[k] += 1
+ messages.append(create_message(k, "%s-%i" % (k, counters[k])))
+ # make sure we have sent at least one message for every key
+ for k, v in counters.iteritems():
+ if v == 0:
+ counters[k] += 1
+ messages.append(create_message(k, "%s-%i" % (k, counters[k])))
+
+ for m in messages:
+ snd.send(m)
+
+ rcv = self.ssn.receiver("lvq; {mode: browse}")
+ retrieved = fetch_all_as_tuples(rcv)
+ print [v for k, v in retrieved]
+
+ for k, v in retrieved:
+ assert v == "%s-%i" % (k, counters[k])
+ assert len(retrieved) <= ring_size
+
+ def test_ring_lvq1(self):
+ self.check_ring_lvq(25, ["a","b","c","d"], 50)
+
+ def test_ring_lvq2(self):
+ self.check_ring_lvq(5, ["a","b","c","d","e","f","g"], 50)
+
+ def test_ring_lvq3(self):
+ self.check_ring_lvq(49, ["a"], 50)
def create_message(key, content):
msg = Message(content=content, properties={"lvq-key":key})
@@ -72,3 +109,14 @@ def fetch_all(rcv):
except Empty:
break
return content
+
+def fetch_all_as_tuples(rcv):
+ content = []
+ while True:
+ try:
+ m = rcv.fetch(0)
+ k = m.properties["lvq-key"]
+ content.append((k, m.content))
+ except Empty:
+ break
+ return content
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
Re: svn commit: r1650196 - in /qpid/trunk/qpid: cpp/src/CMakeLists.txt
cpp/src/qpid/broker/LossyQueue.h cpp/src/qpid/broker/Lvq.h cpp/src/qpid/broker/QueueFactory.cpp
tests/src/py/qpid_tests/broker_0_10/lvq.py
Posted by Gordon Sim <gs...@redhat.com>.
On 01/07/2015 11:28 PM, gsim@apache.org wrote:
> Author: gsim
> Date: Wed Jan 7 23:28:54 2015
> New Revision: 1650196
>
> URL: http://svn.apache.org/r1650196
> Log:
> QPID-6299: allow ring and lvq behaviours to be combined
>
> Modified:
> qpid/trunk/qpid/cpp/src/CMakeLists.txt
> qpid/trunk/qpid/cpp/src/qpid/broker/LossyQueue.h
> qpid/trunk/qpid/cpp/src/qpid/broker/Lvq.h
> qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp
> qpid/trunk/qpid/tests/src/py/qpid_tests/broker_0_10/lvq.py
I broke the build with this commit as I forgot to commit the two new
files. My apologies! They are checked in now.
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org