summaryrefslogtreecommitdiff
path: root/qpid/cpp/src/tests/test_store.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'qpid/cpp/src/tests/test_store.cpp')
-rw-r--r--qpid/cpp/src/tests/test_store.cpp101
1 files changed, 93 insertions, 8 deletions
diff --git a/qpid/cpp/src/tests/test_store.cpp b/qpid/cpp/src/tests/test_store.cpp
index eac4deda2d..fc44889f33 100644
--- a/qpid/cpp/src/tests/test_store.cpp
+++ b/qpid/cpp/src/tests/test_store.cpp
@@ -40,14 +40,19 @@
#include "qpid/sys/Thread.h"
#include "qpid/Plugin.h"
#include "qpid/Options.h"
+#include "qpid/RefCounted.h"
+#include "qpid/Msg.h"
#include <boost/cast.hpp>
#include <boost/lexical_cast.hpp>
#include <memory>
+#include <ostream>
#include <fstream>
+#include <sstream>
-using namespace qpid;
-using namespace broker;
using namespace std;
+using namespace boost;
+using namespace qpid;
+using namespace qpid::broker;
using namespace qpid::sys;
namespace qpid {
@@ -57,11 +62,13 @@ struct TestStoreOptions : public Options {
string name;
string dump;
+ string events;
TestStoreOptions() : Options("Test Store Options") {
addOptions()
("test-store-name", optValue(name, "NAME"), "Name of test store instance.")
("test-store-dump", optValue(dump, "FILE"), "File to dump enqueued messages.")
+ ("test-store-events", optValue(events, "FILE"), "File to log events, 1 line per event.")
;
}
};
@@ -82,24 +89,74 @@ class TestStore : public NullMessageStore {
TestStore(const TestStoreOptions& opts, Broker& broker_)
: options(opts), name(opts.name), broker(broker_)
{
- QPID_LOG(info, "TestStore name=" << name << " dump=" << options.dump);
- if (!options.dump.empty())
+ QPID_LOG(info, "TestStore name=" << name
+ << " dump=" << options.dump
+ << " events=" << options.events);
+
+ if (!options.dump.empty())
dump.reset(new ofstream(options.dump.c_str()));
+ if (!options.events.empty())
+ events.reset(new ofstream(options.events.c_str()));
}
~TestStore() {
for_each(threads.begin(), threads.end(), boost::bind(&Thread::join, _1));
}
- virtual bool isNull() const { return false; }
-
- void enqueue(TransactionContext* ,
+ // Dummy transaction context.
+ struct TxContext : public TPCTransactionContext {
+ static int nextId;
+ string id;
+ TxContext() : id(lexical_cast<string>(nextId++)) {}
+ TxContext(string xid) : id(xid) {}
+ };
+
+ static string getId(const TransactionContext& tx) {
+ const TxContext* tc = dynamic_cast<const TxContext*>(&tx);
+ assert(tc);
+ return tc->id;
+ }
+
+
+ bool isNull() const { return false; }
+
+ void log(const string& msg) {
+ QPID_LOG(info, "test_store: " << msg);
+ if (events.get()) *events << msg << endl << std::flush;
+ }
+
+ auto_ptr<TransactionContext> begin() {
+ auto_ptr<TxContext> tx(new TxContext());
+ log(Msg() << "<begin tx " << tx->id << ">");
+ return auto_ptr<TransactionContext>(tx);
+ }
+
+ auto_ptr<TPCTransactionContext> begin(const std::string& xid) {
+ auto_ptr<TxContext> tx(new TxContext(xid));
+ log(Msg() << "<begin tx " << tx->id << ">");
+ return auto_ptr<TPCTransactionContext>(tx);
+ }
+
+ string getContent(const intrusive_ptr<PersistableMessage>& msg) {
+ intrusive_ptr<broker::Message::Encoding> enc(
+ dynamic_pointer_cast<broker::Message::Encoding>(msg));
+ return enc->getContent();
+ }
+
+ void enqueue(TransactionContext* tx,
const boost::intrusive_ptr<PersistableMessage>& pmsg,
- const PersistableQueue& )
+ const PersistableQueue& queue)
{
+ QPID_LOG(debug, "TestStore enqueue " << queue.getName());
qpid::broker::amqp_0_10::MessageTransfer* msg = dynamic_cast<qpid::broker::amqp_0_10::MessageTransfer*>(pmsg.get());
assert(msg);
+ ostringstream o;
+ o << "<enqueue " << queue.getName() << " " << getContent(msg);
+ if (tx) o << " tx=" << getId(*tx);
+ o << ">";
+ log(o.str());
+
// Dump the message if there is a dump file.
if (dump.get()) {
msg->getFrames().getMethod()->print(*dump);
@@ -144,6 +201,31 @@ class TestStore : public NullMessageStore {
msg->enqueueComplete();
}
+ void dequeue(TransactionContext* tx,
+ const boost::intrusive_ptr<PersistableMessage>& msg,
+ const PersistableQueue& queue)
+ {
+ QPID_LOG(debug, "TestStore dequeue " << queue.getName());
+ ostringstream o;
+ o<< "<dequeue " << queue.getName() << " " << getContent(msg);
+ if (tx) o << " tx=" << getId(*tx);
+ o << ">";
+ log(o.str());
+ }
+
+ void prepare(TPCTransactionContext& txn) {
+ log(Msg() << "<prepare tx=" << getId(txn) << ">");
+ }
+
+ void commit(TransactionContext& txn) {
+ log(Msg() << "<commit tx=" << getId(txn) << ">");
+ }
+
+ void abort(TransactionContext& txn) {
+ log(Msg() << "<abort tx=" << getId(txn) << ">");
+ }
+
+
private:
static const string TEST_STORE_DO, EXCEPTION, EXIT_PROCESS, ASYNC;
TestStoreOptions options;
@@ -151,8 +233,11 @@ class TestStore : public NullMessageStore {
Broker& broker;
vector<Thread> threads;
std::auto_ptr<ofstream> dump;
+ std::auto_ptr<ofstream> events;
};
+int TestStore::TxContext::nextId(1);
+
const string TestStore::TEST_STORE_DO = "TEST_STORE_DO: ";
const string TestStore::EXCEPTION = "exception";
const string TestStore::EXIT_PROCESS = "exit_process";