diff options
Diffstat (limited to 'cpp/src/tests/txtest.cpp')
| -rw-r--r-- | cpp/src/tests/txtest.cpp | 115 |
1 files changed, 66 insertions, 49 deletions
diff --git a/cpp/src/tests/txtest.cpp b/cpp/src/tests/txtest.cpp index f0e70e83e9..d0ba2f1245 100644 --- a/cpp/src/tests/txtest.cpp +++ b/cpp/src/tests/txtest.cpp @@ -7,9 +7,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 @@ -33,12 +33,17 @@ #include "qpid/client/SubscriptionManager.h" #include "qpid/framing/Array.h" #include "qpid/framing/Buffer.h" +#include "qpid/framing/Uuid.h" +#include "qpid/sys/Thread.h" using namespace qpid; using namespace qpid::client; using namespace qpid::sys; using std::string; +namespace qpid { +namespace tests { + typedef std::vector<std::string> StringSet; struct Args : public qpid::TestOptions { @@ -51,13 +56,14 @@ struct Args : public qpid::TestOptions { uint txCount; uint totalMsgCount; bool dtx; + bool quiet; - Args() : init(true), transfer(true), check(true), - size(256), durable(true), queues(2), + Args() : init(true), transfer(true), check(true), + size(256), durable(true), queues(2), base("tx-test"), msgsPerTx(1), txCount(1), totalMsgCount(10), - dtx(false) + dtx(false), quiet(false) { - addOptions() + addOptions() ("init", optValue(init, "yes|no"), "Declare queues and populate one with the initial set of messages.") ("transfer", optValue(transfer, "yes|no"), "'Move' messages from one queue to another using transactions to ensure no message loss.") @@ -69,7 +75,8 @@ struct Args : public qpid::TestOptions { ("messages-per-tx", optValue(msgsPerTx, "N"), "number of messages transferred per transaction") ("tx-count", optValue(txCount, "N"), "number of transactions per 'agent'") ("total-messages", optValue(totalMsgCount, "N"), "total number of messages in 'circulation'") - ("dtx", optValue(dtx, "yes|no"), "use distributed transactions"); + ("dtx", optValue(dtx, "yes|no"), "use distributed transactions") + ("quiet", optValue(quiet), "reduce output from test"); } }; @@ -79,7 +86,7 @@ std::string generateData(uint size) { if (size < chars.length()) { return chars.substr(0, size); - } + } std::string data; for (uint i = 0; i < (size / chars.length()); i++) { data += chars; @@ -99,18 +106,18 @@ void generateSet(const std::string& base, uint count, StringSet& collection) Args opts; -struct Client +struct Client { Connection connection; AsyncSession session; - Client() + Client() { opts.open(connection); session = connection.newSession(); } - ~Client() + ~Client() { try{ session.close(); @@ -126,38 +133,44 @@ struct Transfer : public Client, public Runnable std::string src; std::string dest; Thread thread; - unsigned long xid_cnt; framing::Xid xid; - Transfer(const std::string& to, const std::string& from) : src(to), dest(from), xid_cnt(0), xid(0x4c414e47, "", from) {} + Transfer(const std::string& to, const std::string& from) : src(to), dest(from), xid(0x4c414e47, "", from) {} - void run() + void run() { try { - + if (opts.dtx) session.dtxSelect(); else session.txSelect(); SubscriptionManager subs(session); - - LocalQueue lq(AckPolicy(0));//manual acking - subs.setFlowControl(opts.msgsPerTx, SubscriptionManager::UNLIMITED, true); - subs.subscribe(lq, src); - + + LocalQueue lq; + SubscriptionSettings settings(FlowControl::messageWindow(opts.msgsPerTx)); + settings.autoAck = 0; // Disabled + Subscription sub = subs.subscribe(lq, src, settings); + for (uint t = 0; t < opts.txCount; t++) { Message in; Message out("", dest); if (opts.dtx) { - setNextXid(xid); + setNewXid(xid); session.dtxStart(arg::xid=xid); } for (uint m = 0; m < opts.msgsPerTx; m++) { in = lq.pop(); - out.setData(in.getData()); + std::string& data = in.getData(); + if (data.size() != opts.size) { + std::ostringstream oss; + oss << "Message size incorrect: size=" << in.getData().size() << "; expected " << opts.size; + throw std::runtime_error(oss.str()); + } + out.setData(data); out.getMessageProperties().setCorrelationId(in.getMessageProperties().getCorrelationId()); out.getDeliveryProperties().setDeliveryMode(in.getDeliveryProperties().getDeliveryMode()); session.messageTransfer(arg::content=out, arg::acceptMode=1); } - lq.getAckPolicy().ackOutstanding(session); + sub.accept(sub.getUnaccepted()); if (opts.dtx) { session.dtxEnd(arg::xid=xid); session.dtxPrepare(arg::xid=xid); @@ -171,14 +184,13 @@ struct Transfer : public Client, public Runnable } } - void setNextXid(framing::Xid& xid) { - std::ostringstream oss; - oss << std::setfill('0') << std::hex << "xid-" << std::setw(12) << (++xid_cnt); - xid.setGlobalId(oss.str()); + void setNewXid(framing::Xid& xid) { + framing::Uuid uuid(true); + xid.setGlobalId(uuid.str()); } }; -struct Controller : public Client +struct Controller : public Client { StringSet ids; StringSet queues; @@ -189,7 +201,7 @@ struct Controller : public Client generateSet("msg", opts.totalMsgCount, ids); } - void init() + void init() { //declare queues for (StringSet::iterator i = queues.begin(); i != queues.end(); i++) { @@ -217,7 +229,7 @@ struct Controller : public Client StringSet::iterator next = i + 1; if (next == queues.end()) next = queues.begin(); - std::cout << "Transfering from " << *i << " to " << *next << std::endl; + if (!opts.quiet) std::cout << "Transfering from " << *i << " to " << *next << std::endl; agents.push_back(new Transfer(*i, *next)); agents.back().thread = Thread(agents.back()); } @@ -227,11 +239,9 @@ struct Controller : public Client } } - void check() + int check() { SubscriptionManager subs(session); - subs.setFlowControl(SubscriptionManager::UNLIMITED, SubscriptionManager::UNLIMITED, false); - subs.setAcceptMode(1/*not-required*/); // Recover DTX transactions (if any) if (opts.dtx) { @@ -241,13 +251,13 @@ struct Controller : public Client xidArr.collect(inDoubtXids); if (inDoubtXids.size()) { - std::cout << "Recovering DTX in-doubt transaction(s):" << std::endl; + if (!opts.quiet) std::cout << "Recovering DTX in-doubt transaction(s):" << std::endl; framing::StructHelper decoder; framing::Xid xid; // abort even, commit odd transactions for (unsigned i = 0; i < inDoubtXids.size(); i++) { decoder.decode(xid, inDoubtXids[i]); - std::cout << (i%2 ? " * aborting " : " * committing "); + if (!opts.quiet) std::cout << (i%2 ? " * aborting " : " * committing "); xid.print(std::cout); std::cout << std::endl; if (i%2) { @@ -262,9 +272,10 @@ struct Controller : public Client StringSet drained; //drain each queue and verify the correct set of messages are available for (StringSet::iterator i = queues.begin(); i != queues.end(); i++) { - //subscribe, allocate credit and flush - LocalQueue lq(AckPolicy(0));//manual acking - subs.subscribe(lq, *i, *i); + //subscribe, allocate credit and flushn + LocalQueue lq; + SubscriptionSettings settings(FlowControl::unlimited(), ACCEPT_MODE_NONE); + subs.subscribe(lq, *i, settings); session.messageFlush(arg::destination=*i); session.sync(); @@ -275,7 +286,7 @@ struct Controller : public Client drained.push_back(m.getMessageProperties().getCorrelationId()); ++count; } - std::cout << "Drained " << count << " messages from " << *i << std::endl; + if (!opts.quiet) std::cout << "Drained " << count << " messages from " << *i << std::endl; } sort(ids.begin(), ids.end()); @@ -283,41 +294,47 @@ struct Controller : public Client //check that drained == ids StringSet missing; - set_difference(ids.begin(), ids.end(), drained.begin(), drained.end(), back_inserter(missing)); + set_difference(ids.begin(), ids.end(), drained.begin(), drained.end(), back_inserter(missing)); StringSet extra; - set_difference(drained.begin(), drained.end(), ids.begin(), ids.end(), back_inserter(extra)); + set_difference(drained.begin(), drained.end(), ids.begin(), ids.end(), back_inserter(extra)); if (missing.empty() && extra.empty()) { - std::cout << "All expected messages were retrieved." << std::endl; + std::cout << "All expected messages were retrieved." << std::endl; + return 0; } else { if (!missing.empty()) { std::cout << "The following ids were missing:" << std::endl; for (StringSet::iterator i = missing.begin(); i != missing.end(); i++) { - std::cout << " '" << *i << "'" << std::endl; - } + std::cout << " '" << *i << "'" << std::endl; + } } if (!extra.empty()) { std::cout << "The following extra ids were encountered:" << std::endl; for (StringSet::iterator i = extra.begin(); i != extra.end(); i++) { - std::cout << " '" << *i << "'" << std::endl; - } + std::cout << " '" << *i << "'" << std::endl; + } } + return 1; } } }; +}} // namespace qpid::tests + +using namespace qpid::tests; + int main(int argc, char** argv) { try { opts.parse(argc, argv); Controller controller; - if (opts.init) controller.init(); + if (opts.init) controller.init(); if (opts.transfer) controller.transfer(); - if (opts.check) controller.check(); + if (opts.check) return controller.check(); return 0; } catch(const std::exception& e) { std::cout << e.what() << std::endl; } - return 1; + return 2; } |