Author: cctrieloff
Date: Thu Jul 30 20:30:18 2009
New Revision: 799435
URL: http://svn.apache.org/viewvc?rev=799435&view=rev
Log:
Make replication of queue events not dependant on abs queue position, but
rather use source queue sequence numbers. Also add in more error checking and
tests
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
qpid/trunk/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp
qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp
qpid/trunk/qpid/cpp/src/qpid/replication/constants.h
qpid/trunk/qpid/cpp/src/tests/replication_test
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=799435&r1=799434&r2=799435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Thu Jul 30 20:30:18 2009
@@ -1012,6 +1012,10 @@
sequence = n;
}
+SequenceNumber Queue::getPosition() {
+ return sequence;
+}
+
int Queue::getEventMode() { return eventMode; }
void Queue::setQueueEventManager(QueueEvents& mgr)
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=799435&r1=799434&r2=799435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Thu Jul 30 20:30:18 2009
@@ -324,6 +324,9 @@
* Used by cluster to replicate queues.
*/
void setPosition(framing::SequenceNumber pos);
+ /** return current position sequence number for the next message
on the queue.
+ */
+ framing::SequenceNumber getPosition();
int getEventMode();
void setQueueEventManager(QueueEvents&);
QPID_BROKER_EXTERN void insertSequenceNumbers(const std::string&
key);
Modified: qpid/trunk/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp?rev=799435&r1=799434&r2=799435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp
(original)
+++ qpid/trunk/qpid/cpp/src/qpid/replication/ReplicatingEventListener.cpp Thu
Jul 30 20:30:18 2009
@@ -72,6 +72,7 @@
FieldTable& headers =
msg->getProperties<MessageProperties>()->getApplicationHeaders();
headers.setString(REPLICATION_TARGET_QUEUE, enqueued.queue->getName());
headers.setInt(REPLICATION_EVENT_TYPE, ENQUEUE);
+ headers.setInt(QUEUE_MESSAGE_POSITION,enqueued.position);
route(msg);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp?rev=799435&r1=799434&r2=799435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp Thu Jul 30
20:30:18 2009
@@ -83,15 +83,27 @@
std::string queueName = args->getAsString(REPLICATION_TARGET_QUEUE);
Queue::shared_ptr queue = queues.find(queueName);
if (queue) {
- FieldTable& headers =
msg.getMessage().getProperties<MessageProperties>()->getApplicationHeaders();
- headers.erase(REPLICATION_TARGET_QUEUE);
- headers.erase(REPLICATION_EVENT_SEQNO);
- headers.erase(REPLICATION_EVENT_TYPE);
- msg.deliverTo(queue);
- QPID_LOG(debug, "Enqueued replicated message onto " << queueName);
- if (mgmtExchange != 0) {
- mgmtExchange->inc_msgRoutes();
- mgmtExchange->inc_byteRoutes( msg.contentSize());
+
+ SequenceNumber seqno1(args->getAsInt(QUEUE_MESSAGE_POSITION));
+
+ if (queue->getPosition() > seqno1) // test queue.pos < seqnumber
+ {
+ QPID_LOG(error, "Cannot enqueue replicated message. Destination
Queue " << queueName << " ahead of source queue");
+ mgmtExchange->inc_msgDrops();
+ mgmtExchange->inc_byteDrops(msg.contentSize());
+ } else {
+ queue->setPosition(--seqno1); // note that queue will ++ before
enqueue.
+
+ FieldTable& headers =
msg.getMessage().getProperties<MessageProperties>()->getApplicationHeaders();
+ headers.erase(REPLICATION_TARGET_QUEUE);
+ headers.erase(REPLICATION_EVENT_SEQNO);
+ headers.erase(REPLICATION_EVENT_TYPE);
+ msg.deliverTo(queue);
+ QPID_LOG(debug, "Enqueued replicated message onto " << queueName);
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_msgRoutes();
+ mgmtExchange->inc_byteRoutes( msg.contentSize());
+ }
}
} else {
QPID_LOG(error, "Cannot enqueue replicated message. Queue " <<
queueName << " does not exist");
Modified: qpid/trunk/qpid/cpp/src/qpid/replication/constants.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/replication/constants.h?rev=799435&r1=799434&r2=799435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/replication/constants.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/replication/constants.h Thu Jul 30 20:30:18
2009
@@ -26,6 +26,7 @@
const std::string REPLICATION_EVENT_SEQNO("qpid.replication.seqno");
const std::string REPLICATION_TARGET_QUEUE("qpid.replication.target_queue");
const std::string DEQUEUED_MESSAGE_POSITION("qpid.replication.message");
+const std::string QUEUE_MESSAGE_POSITION("qpid.replication.queue.position");
const int ENQUEUE(1);
const int DEQUEUE(2);
Modified: qpid/trunk/qpid/cpp/src/tests/replication_test
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/replication_test?rev=799435&r1=799434&r2=799435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/replication_test (original)
+++ qpid/trunk/qpid/cpp/src/tests/replication_test Thu Jul 30 20:30:18 2009
@@ -98,7 +98,6 @@
./receiver --port $BROKER_B --queue queue-b > queue-b-backup.repl
./receiver --port $BROKER_B --queue queue-c > queue-c-backup.repl
- stop_brokers
tail -5 queue-a-input.repl > queue-a-expected.repl
tail -10 queue-b-input.repl > queue-b-expected.repl
@@ -108,6 +107,54 @@
grep 'queue-d does not exist' replication-dest.log > /dev/null || echo
"WARNING: Expected error to be logged!"
+ stop_brokers
+
+ # now check offsets working (enqueue based on position being set, not
queue abs position)
+
+ ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no
--load-module ../.libs/replicating_listener.so --replication-queue replication
--create-replication-queue true --log-enable info+ --log-to-file
replication-source.log --log-to-stderr 0 > qpidd.port
+ BROKER_A=`cat qpidd.port`
+
+ ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no
--load-module ../.libs/replication_exchange.so --log-enable info+
--log-to-file replication-dest.log --log-to-stderr 0 > qpidd.port
+ BROKER_B=`cat qpidd.port`
+
+ $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add exchange
replication replication
+ $PYTHON_DIR/commands/qpid-route --ack 5 queue add "localhost:$BROKER_B"
"localhost:$BROKER_A" replication replication
+
+ $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_A" add queue
queue-e --generate-queue-events 2
+ $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add queue queue-e
+ $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_A" add queue
queue-d --generate-queue-events 2
+
+ i=1
+ while [ $i -le 10 ]; do
+ echo Message $i for A >> queue-e-input.repl
+ i=`expr $i + 1`
+ done
+
+ ./sender --port $BROKER_A --routing-key queue-e --send-eos 1 <
queue-e-input.repl
+ ./receiver --port $BROKER_A --queue queue-e --messages 10 > /dev/null
+ # now check offsets working
+
+ ../qpidd -q --port $BROKER_B
+ unset BROKER_B
+ ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no
--load-module ../.libs/replication_exchange.so --log-enable info+
--log-to-file replication-dest.log --log-to-stderr 0 > qpidd.port
+ BROKER_B=`cat qpidd.port`
+
+ $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add queue queue-e
+ $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add exchange
replication replication
+ $PYTHON_DIR/commands/qpid-route --ack 5 queue add "localhost:$BROKER_B"
"localhost:$BROKER_A" replication replication
+ # now send another 15
+ i=11
+ while [ $i -le 15 ]; do
+ echo Message $i for A >> queue-e1-input.repl
+ i=`expr $i + 1`
+ done
+ ./sender --port $BROKER_A --routing-key queue-e --send-eos 1 <
queue-e1-input.repl
+
+ ./receiver --port $BROKER_B --queue queue-e > queue-e-backup.repl
+ diff queue-e-backup.repl queue-e1-input.repl || FAIL=1
+
+ stop_brokers
+
if [ x$FAIL != x ]; then
echo replication test failed: expectations not met!
exit 1
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
