You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2009/05/27 17:12:33 UTC
svn commit: r779205 - /qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
Author: kpvdr
Date: Wed May 27 15:12:32 2009
New Revision: 779205
URL: http://svn.apache.org/viewvc?rev=779205&view=rev
Log:
Fixed erroneous use of arg::durable in messageTransfer()
Modified:
qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
Modified: qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp?rev=779205&r1=779204&r2=779205&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_test.cpp Wed May 27 15:12:32 2009
@@ -131,9 +131,16 @@
return m.getMessageProperties().getApplicationHeaders().getAsInt64("qpid.msg_sequence");
}
-Message ttlMessage(const string& data, const string& key, uint64_t ttl) {
+Message ttlMessage(const string& data, const string& key, uint64_t ttl, bool durable = false) {
Message m(data, key);
m.getDeliveryProperties().setTtl(ttl);
+ if (durable) m.getDeliveryProperties().setDeliveryMode(framing::PERSISTENT);
+ return m;
+}
+
+Message makeMessage(const string& data, const string& key, bool durable = false) {
+ Message m(data, key);
+ if (durable) m.getDeliveryProperties().setDeliveryMode(framing::PERSISTENT);
return m;
}
@@ -220,10 +227,10 @@
Client c1(cluster[1], "c1");
c0.session.queueDeclare("p", arg::durable=durableFlag);
c0.session.queueDeclare("q", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=ttlMessage("a", "q", 200), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("b", "q"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=ttlMessage("x", "p", 10000), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("y", "p"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=ttlMessage("a", "q", 200, durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("b", "q", durableFlag));
+ c0.session.messageTransfer(arg::content=ttlMessage("x", "p", 10000, durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("y", "p", durableFlag));
cluster.add();
Client c2(cluster[1], "c2");
@@ -248,14 +255,14 @@
c0.session.queueDeclare(arg::queue="q", arg::durable=durableFlag);
c0.session.exchangeDeclare(arg::exchange="ex", arg::type="direct", arg::arguments=ftargs);
c0.session.exchangeBind(arg::exchange="ex", arg::queue="q", arg::bindingKey="k");
- c0.session.messageTransfer(arg::content=Message("1", "k"), arg::destination="ex", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("2", "k"), arg::destination="ex", arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("1", "k", durableFlag), arg::destination="ex");
+ c0.session.messageTransfer(arg::content=makeMessage("2", "k", durableFlag), arg::destination="ex");
BOOST_CHECK_EQUAL(1, getMsgSequence(c0.subs.get("q", TIMEOUT)));
BOOST_CHECK_EQUAL(2, getMsgSequence(c0.subs.get("q", TIMEOUT)));
cluster.add();
Client c1(cluster[1]);
- c1.session.messageTransfer(arg::content=Message("3", "k"), arg::destination="ex", arg::durable=durableFlag);
+ c1.session.messageTransfer(arg::content=makeMessage("3", "k", durableFlag), arg::destination="ex");
BOOST_CHECK_EQUAL(3, getMsgSequence(c1.subs.get("q", TIMEOUT)));
}
@@ -265,22 +272,22 @@
ClusterFixture cluster(1, args, -1);
Client c0(cluster[0], "c0");
c0.session.queueDeclare(arg::queue="q", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("A", "q"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("B", "q"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("A", "q", durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("B", "q", durableFlag));
// Start a transaction that will commit.
Session commitSession = c0.connection.newSession("commit");
SubscriptionManager commitSubs(commitSession);
commitSession.txSelect();
- commitSession.messageTransfer(arg::content=Message("a", "q"), arg::durable=durableFlag);
- commitSession.messageTransfer(arg::content=Message("b", "q"), arg::durable=durableFlag);
+ commitSession.messageTransfer(arg::content=makeMessage("a", "q", durableFlag));
+ commitSession.messageTransfer(arg::content=makeMessage("b", "q", durableFlag));
BOOST_CHECK_EQUAL(commitSubs.get("q", TIMEOUT).getData(), "A");
// Start a transaction that will roll back.
Session rollbackSession = c0.connection.newSession("rollback");
SubscriptionManager rollbackSubs(rollbackSession);
rollbackSession.txSelect();
- rollbackSession.messageTransfer(arg::content=Message("1", "q"), arg::durable=durableFlag);
+ rollbackSession.messageTransfer(arg::content=makeMessage("1", "q", durableFlag));
Message rollbackMessage = rollbackSubs.get("q", TIMEOUT);
BOOST_CHECK_EQUAL(rollbackMessage.getData(), "B");
@@ -291,9 +298,9 @@
// More transactional work
BOOST_CHECK_EQUAL(c1.session.queueQuery("q").getMessageCount(), 0u);
- rollbackSession.messageTransfer(arg::content=Message("2", "q"), arg::durable=durableFlag);
- commitSession.messageTransfer(arg::content=Message("c", "q"), arg::durable=durableFlag);
- rollbackSession.messageTransfer(arg::content=Message("3", "q"), arg::durable=durableFlag);
+ rollbackSession.messageTransfer(arg::content=makeMessage("2", "q", durableFlag));
+ commitSession.messageTransfer(arg::content=makeMessage("c", "q", durableFlag));
+ rollbackSession.messageTransfer(arg::content=makeMessage("3", "q", durableFlag));
BOOST_CHECK_EQUAL(c1.session.queueQuery("q").getMessageCount(), 0u);
@@ -323,7 +330,7 @@
// Create unacked message: acquired but not accepted.
SubscriptionSettings manualAccept(FlowControl::unlimited(), ACCEPT_MODE_EXPLICIT, ACQUIRE_MODE_PRE_ACQUIRED, 0);
c0.session.queueDeclare("q1", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("11","q1"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("11","q1", durableFlag));
LocalQueue q1;
c0.subs.subscribe(q1, "q1", manualAccept);
BOOST_CHECK_EQUAL(q1.get(TIMEOUT).getData(), "11"); // Acquired but not accepted
@@ -332,8 +339,8 @@
// Create unacked message: not acquired, accepted or completeed.
SubscriptionSettings manualAcquire(FlowControl::unlimited(), ACCEPT_MODE_EXPLICIT, ACQUIRE_MODE_NOT_ACQUIRED, 0);
c0.session.queueDeclare("q2", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("21","q2"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("22","q2"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("21","q2", durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("22","q2", durableFlag));
LocalQueue q2;
c0.subs.subscribe(q2, "q2", manualAcquire);
m = q2.get(TIMEOUT); // Not acquired or accepted, still on queue
@@ -347,8 +354,8 @@
// Create empty credit record: acquire and accept but don't complete.
SubscriptionSettings manualComplete(FlowControl::messageWindow(1), ACCEPT_MODE_EXPLICIT, ACQUIRE_MODE_PRE_ACQUIRED, 1, MANUAL_COMPLETION);
c0.session.queueDeclare("q3", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("31", "q3"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("32", "q3"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("31", "q3", durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("32", "q3", durableFlag));
LocalQueue q3;
c0.subs.subscribe(q3, "q3", manualComplete);
Message m31=q3.get(TIMEOUT);
@@ -391,8 +398,8 @@
// Do work in a transaction.
c0.session.txSelect();
c0.session.queueDeclare("q", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("1","q"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("2","q"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("1","q", durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("2","q", durableFlag));
Message m;
BOOST_CHECK(c0.subs.get(m, "q", TIMEOUT));
BOOST_CHECK_EQUAL(m.getData(), "1");
@@ -409,7 +416,7 @@
BOOST_CHECK_EQUAL(m.getData(), "2");
// Another transaction with both members active.
- c0.session.messageTransfer(arg::content=Message("3","q"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("3","q", durableFlag));
BOOST_CHECK_EQUAL(c1.session.queueQuery(arg::queue="q").getMessageCount(), 0u);
c0.session.txCommit();
BOOST_CHECK_EQUAL(c1.session.queueQuery(arg::queue="q").getMessageCount(), 1u);
@@ -511,10 +518,10 @@
Client c2(cluster[2], "c2");
// Transfer messages
- c0.session.messageTransfer(arg::content=Message("aaa", "q"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("aaa", "q", durableFlag));
- c0.session.messageTransfer(arg::content=Message("bbb", "p"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("ccc", "p"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("bbb", "p", durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("ccc", "p", durableFlag));
// Activate the subscription, ensure message removed on all queues.
c0.subs.setFlowControl("q", FlowControl::unlimited());
@@ -539,7 +546,7 @@
cluster.killWithSilencer(0,c0.connection,9);
BOOST_REQUIRE_EQUAL(knownBrokerPorts(c1.connection, 2).size(), 2u);
for (int i = 0; i < 10; ++i) {
- c1.session.messageTransfer(arg::content=Message("xxx", "q"), arg::durable=durableFlag);
+ c1.session.messageTransfer(arg::content=makeMessage("xxx", "q", durableFlag));
BOOST_REQUIRE(c1.subs.get(m, "q", TIMEOUT));
BOOST_REQUIRE_EQUAL(m.getData(), "xxx");
}
@@ -553,8 +560,8 @@
// Create some shared state.
c0.session.queueDeclare("q", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("foo","q"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("bar","q"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("foo","q", durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("bar","q", durableFlag));
while (c0.session.queueQuery("q").getMessageCount() != 2)
sys::usleep(1000); // Wait for message to show up on broker 0.
@@ -564,11 +571,11 @@
// Do some work post-add
c0.session.queueDeclare("p", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("pfoo","p"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("pfoo","p", durableFlag));
// Do some work post-join
BOOST_REQUIRE_EQUAL(knownBrokerPorts(c0.connection, 2).size(), 2u);
- c0.session.messageTransfer(arg::content=Message("pbar","p"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("pbar","p", durableFlag));
// Verify new brokers have state.
Message m;
@@ -619,8 +626,8 @@
ClusterFixture cluster(2, args, -1);
Client c0(cluster[0]);
c0.session.queueDeclare("q", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("foo", "q"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("bar", "q"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("foo", "q", durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("bar", "q", durableFlag));
c0.session.close();
Client c1(cluster[1]);
Message msg;
@@ -637,8 +644,8 @@
ClusterFixture cluster(3, args, -1);
Client c0(cluster[0], "c0");
c0.session.queueDeclare("q", arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("foo", "q"), arg::durable=durableFlag);
- c0.session.messageTransfer(arg::content=Message("bar", "q"), arg::durable=durableFlag);
+ c0.session.messageTransfer(arg::content=makeMessage("foo", "q", durableFlag));
+ c0.session.messageTransfer(arg::content=makeMessage("bar", "q", durableFlag));
Message msg;
@@ -670,8 +677,8 @@
// Now send messages
Client c1(cluster[1]);
- c1.session.messageTransfer(arg::content=Message("foo", "q"), arg::durable=durableFlag);
- c1.session.messageTransfer(arg::content=Message("bar", "q"), arg::durable=durableFlag);
+ c1.session.messageTransfer(arg::content=makeMessage("foo", "q", durableFlag));
+ c1.session.messageTransfer(arg::content=makeMessage("bar", "q", durableFlag));
// Check they arrived
Message m;
@@ -698,7 +705,7 @@
void execute(AsyncSession& session, bool)
{
- session.messageTransfer(arg::content=Message(content, queue), arg::durable=durableFlag);
+ session.messageTransfer(arg::content=makeMessage(content, queue, durableFlag));
}
};
@@ -783,12 +790,12 @@
QueueOptions options;
options.setSizePolicy(REJECT, 0, 2);
c1.session.queueDeclare("q", arg::arguments=options, arg::durable=durableFlag);
- c1.session.messageTransfer(arg::content=Message("one", "q"));
+ c1.session.messageTransfer(arg::content=makeMessage("one", "q", durableFlag));
cluster.add();
Client c2(cluster[1], "c2");
- c2.session.messageTransfer(arg::content=Message("two", "q"));
-
- BOOST_CHECK_THROW(c2.session.messageTransfer(arg::content=Message("three", "q")), framing::ResourceLimitExceededException);
+ c2.session.messageTransfer(arg::content=makeMessage("two", "q", durableFlag));
+
+ BOOST_CHECK_THROW(c2.session.messageTransfer(arg::content=makeMessage("three", "q", durableFlag)), framing::ResourceLimitExceededException);
Message received;
BOOST_CHECK(c1.subs.get(received, "q"));
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org