summaryrefslogtreecommitdiff
path: root/qpid/cpp/src
diff options
context:
space:
mode:
Diffstat (limited to 'qpid/cpp/src')
-rw-r--r--qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp2
-rw-r--r--qpid/cpp/src/qpid/linearstore/ISSUES30
-rw-r--r--qpid/cpp/src/qpid/linearstore/journal/jdir.cpp152
-rw-r--r--qpid/cpp/src/qpid/linearstore/journal/jdir.h2
-rw-r--r--qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp2
-rw-r--r--qpid/cpp/src/qpid/messaging/exceptions.cpp2
-rw-r--r--qpid/cpp/src/tests/legacystore/CMakeLists.txt24
-rw-r--r--qpid/cpp/src/tests/legacystore/OrderingTest.cpp171
-rw-r--r--qpid/cpp/src/tests/legacystore/SimpleTest.cpp500
-rw-r--r--qpid/cpp/src/tests/legacystore/TransactionalTest.cpp354
-rw-r--r--qpid/cpp/src/tests/legacystore/TwoPhaseCommitTest.cpp678
-rwxr-xr-xqpid/cpp/src/tests/linearstore/linearstoredirsetup.sh51
-rwxr-xr-xqpid/cpp/src/tests/linearstore/tx-test-soak.sh31
13 files changed, 147 insertions, 1852 deletions
diff --git a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
index e43abb1a56..057e752c90 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
+++ b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
@@ -368,6 +368,8 @@ bool SessionImpl::nextReceiver(qpid::messaging::Receiver& receiver, qpid::messag
throw qpid::messaging::UnauthorizedAccess(e.what());
} catch (const qpid::SessionException& e) {
throw qpid::messaging::SessionError(e.what());
+ } catch (const qpid::ClosedException&) {
+ throw qpid::messaging::SessionClosed();
} catch (const qpid::ConnectionException& e) {
throw qpid::messaging::ConnectionError(e.what());
} catch (const qpid::ChannelException& e) {
diff --git a/qpid/cpp/src/qpid/linearstore/ISSUES b/qpid/cpp/src/qpid/linearstore/ISSUES
index a9908e882e..ccadefc20c 100644
--- a/qpid/cpp/src/qpid/linearstore/ISSUES
+++ b/qpid/cpp/src/qpid/linearstore/ISSUES
@@ -47,8 +47,6 @@ Current/pending:
svn r.1558592 2014-01-15 fixes an issue with using /dev/random as a source of random numbers for Journal serial numbers.
svn r.1558913 2014-01-16 replaces use of /dev/urandom with several calls to rand() to construct a 64-bit random number.
* Recommend rebuilding and testing for performance again with these two fixes. Marked POST.
-# - 1036026 [LinearStore] Qpid linear store unable to create durable queue - framing-error: Queue <q-name>: create() failed: jexception 0x0000
- UNABLE TO REPRODUCE - but Frantizek has additional info
- 1039522 Qpid crashes while recovering from linear store around apid::linearstore::journal::JournalFile::getFqFileName() including enq_rec::decode() threw JERR_JREC_BAD_RECTAIL
* Possible dup of 1039525
* May be fixed by QPID-5483 - waiting for needinfo, recommend rebuilding with QPID-5483 fix and re-testing. Marked POST.
@@ -56,18 +54,6 @@ Current/pending:
* Possible dup of 1039522
* May be fixed by QPID-5483 - waiting for needinfo, recommend rebuilding with QPID-5483 fix and re-testing. Marked POST.
# - 1049870 [LinearStore] auto-delete property does not survive restart
-# 5480 1053749 [linearstore] Recovery of store failure with "JERR_MAP_NOTFOUND: Key not found in map." error message
- svn r.1564877 2014-02-05: Proposed fix
- * Probability: 6 of 600 (1.0%) using tx-test-soak.sh
- * If broker is started a second time after failure, it starts correctly and test completes ok.
- * Problem: File is being recycled to EFP with still-locked enqueues in it (ie dequeued transactionally).
- * Problem: Record alignment check writes filler records to wrong file when decoding bad record moves across a file boundary
- * Test of fix failed on RHEL-7
-# - 1064181 [linearstore] Qpidd closes transactional client session&connection with async_dequeue() failed
- * jexception 0x010b LinearFileController::getCurrentSerial() threw JERR_NULL
-# - 1064230 [linearstore] Qpidd linearstore recovery sometimes fail to recover messages with recoverMessages() failed
- * jexception 0x0701 RecoveryManager::readNextRemainingRecord() threw JERR_JREC_BADRECTAIL
- * possible dup of 1063700
Fixed/closed (in commit order):
===============================
@@ -104,9 +90,24 @@ NO-JIRA - Added missing Apache copyright/license text
5479 1053701 [linearstore] Using recovered store results in "JERR_JNLF_FILEOFFSOVFL: Attempted to increase submitted offset past file size. (JournalFile::submittedDblkCount)" error message
* Probability: 2 of 600 (0.3%) using tx-test-soak.sh
* Fixed by checkin for QPID-5480, no longer able to reproduce. VERIFIED
+ 5480 1053749 [linearstore] Recovery of store failure with "JERR_MAP_NOTFOUND: Key not found in map." error message
+ svn r.1564877 2014-02-05: Proposed fix
+ * Probability: 6 of 600 (1.0%) using tx-test-soak.sh
+ * If broker is started a second time after failure, it starts correctly and test completes ok.
+ * Problem: File is being recycled to EFP with still-locked enqueues in it (ie dequeued transactionally).
+ * Problem: Record alignment check writes filler records to wrong file when decoding bad record moves across a file boundary
5603 1063700 [linearstore] broker restart fails under stress test
svn r.1574513 2014-03-05: Proposed fix. POST
* jexception 0x0701 RecoveryManager::readNextRemainingRecord() threw JERR_JREC_BADRECTAIL
+ 5607 1064181 [linearstore] Qpidd closes transactional client session&connection with async_dequeue() failed
+ svn r.1575009 2014-03-06 Proposed fix. POST
+ * jexception 0x010b LinearFileController::getCurrentSerial() threw JERR_NULL
+ - 1064230 [linearstore] Qpidd linearstore recovery sometimes fail to recover messages with recoverMessages() failed
+ * jexception 0x0701 RecoveryManager::readNextRemainingRecord() threw JERR_JREC_BADRECTAIL
+ * possible dup of 1063700
+ - 1036026 [LinearStore] Qpid linear store unable to create durable queue - framing-error: Queue <q-name>: create() failed: jexception 0x0000
+ * UNABLE TO REPRODUCE - but Frantizek has additional info
+ * Retested after checkin 1575009, problem solved. VERIFIED
Ordered checkin list:
=====================
@@ -135,6 +136,7 @@ no. svn r Q-JIRA RHBZ Date
19. 1564893 5361 - 2014-02-05
20. 1564935 5361 - 2014-02-05
21. 1574513 5603 1063700 2014-03-05
+22. 1575009 5607 1064181 2014-03-06
See above sections for details on these checkins.
diff --git a/qpid/cpp/src/qpid/linearstore/journal/jdir.cpp b/qpid/cpp/src/qpid/linearstore/journal/jdir.cpp
index 896f44ceff..36f180c21f 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/jdir.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/jdir.cpp
@@ -101,17 +101,9 @@ jdir::clear_dir(const std::string& dirname/*, const std::string&
*/
, const bool create_flag)
{
- DIR* dir = ::opendir(dirname.c_str());
- if (!dir)
- {
- if (errno == 2 && create_flag) // ENOENT (No such file or dir)
- {
- create_dir(dirname);
- return;
- }
- std::ostringstream oss;
- oss << "dir=\"" << dirname << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_JDIR_OPENDIR, oss.str(), "jdir", "clear_dir");
+ DIR* dir = open_dir(dirname, "clear_dir", true);
+ if (!dir && create_flag) {
+ create_dir(dirname);
}
//#ifndef RHM_JOWRITE
struct dirent* entry;
@@ -161,13 +153,7 @@ jdir::push_down(const std::string& dirname, const std::string& target_dir/*, con
{
std::string bak_dir_name = create_bak_dir(dirname/*, bak_dir_base*/);
- DIR* dir = ::opendir(dirname.c_str());
- if (!dir)
- {
- std::ostringstream oss;
- oss << "dir=\"" << dirname << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_JDIR_OPENDIR, oss.str(), "jdir", "push_down");
- }
+ DIR* dir = open_dir(dirname, "push_down", false);
// Copy contents of targetDirName into bak dir
struct dirent* entry;
while ((entry = ::readdir(dir)) != 0)
@@ -251,60 +237,49 @@ jdir::delete_dir(const std::string& dirname, bool children_only)
{
struct dirent* entry;
struct stat s;
- DIR* dir = ::opendir(dirname.c_str());
- if (!dir)
- {
- if (errno == ENOENT) // dir does not exist.
- return;
-
- std::ostringstream oss;
- oss << "dir=\"" << dirname << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_JDIR_OPENDIR, oss.str(), "jdir", "delete_dir");
- }
- else
+ DIR* dir = open_dir(dirname, "delete_dir", true); // true = allow dir does not exist, return 0
+ if (!dir) return;
+ while ((entry = ::readdir(dir)) != 0)
{
- while ((entry = ::readdir(dir)) != 0)
+ // Ignore . and ..
+ if (std::strcmp(entry->d_name, ".") != 0 && std::strcmp(entry->d_name, "..") != 0)
{
- // Ignore . and ..
- if (std::strcmp(entry->d_name, ".") != 0 && std::strcmp(entry->d_name, "..") != 0)
+ std::string full_name(dirname + "/" + entry->d_name);
+ if (::lstat(full_name.c_str(), &s))
{
- std::string full_name(dirname + "/" + entry->d_name);
- if (::lstat(full_name.c_str(), &s))
- {
- ::closedir(dir);
- std::ostringstream oss;
- oss << "stat: file=\"" << full_name << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_JDIR_STAT, oss.str(), "jdir", "delete_dir");
- }
- if (S_ISREG(s.st_mode) || S_ISLNK(s.st_mode)) // This is a file or slink
- {
- if(::unlink(full_name.c_str()))
- {
- ::closedir(dir);
- std::ostringstream oss;
- oss << "unlink: file=\"" << entry->d_name << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_JDIR_UNLINK, oss.str(), "jdir", "delete_dir");
- }
- }
- else if (S_ISDIR(s.st_mode)) // This is a dir
- {
- delete_dir(full_name);
- }
- else // all other types, throw up!
+ ::closedir(dir);
+ std::ostringstream oss;
+ oss << "stat: file=\"" << full_name << "\"" << FORMAT_SYSERR(errno);
+ throw jexception(jerrno::JERR_JDIR_STAT, oss.str(), "jdir", "delete_dir");
+ }
+ if (S_ISREG(s.st_mode) || S_ISLNK(s.st_mode)) // This is a file or slink
+ {
+ if(::unlink(full_name.c_str()))
{
::closedir(dir);
std::ostringstream oss;
- oss << "file=\"" << entry->d_name << "\" is not a dir, file or slink.";
- oss << " (mode=0x" << std::hex << s.st_mode << std::dec << ")";
- throw jexception(jerrno::JERR_JDIR_BADFTYPE, oss.str(), "jdir", "delete_dir");
+ oss << "unlink: file=\"" << entry->d_name << "\"" << FORMAT_SYSERR(errno);
+ throw jexception(jerrno::JERR_JDIR_UNLINK, oss.str(), "jdir", "delete_dir");
}
}
+ else if (S_ISDIR(s.st_mode)) // This is a dir
+ {
+ delete_dir(full_name);
+ }
+ else // all other types, throw up!
+ {
+ ::closedir(dir);
+ std::ostringstream oss;
+ oss << "file=\"" << entry->d_name << "\" is not a dir, file or slink.";
+ oss << " (mode=0x" << std::hex << s.st_mode << std::dec << ")";
+ throw jexception(jerrno::JERR_JDIR_BADFTYPE, oss.str(), "jdir", "delete_dir");
+ }
}
+ }
// FIXME: Find out why this fails with false alarms/errors from time to time...
// While commented out, there is no error capture from reading dir entries.
// check_err(errno, dir, dirname, "delete_dir");
- }
// Now dir is empty, close and delete it
close_dir(dir, dirname, "delete_dir");
@@ -321,14 +296,8 @@ jdir::delete_dir(const std::string& dirname, bool children_only)
std::string
jdir::create_bak_dir(const std::string& dirname)
{
- DIR* dir = ::opendir(dirname.c_str());
+ DIR* dir = open_dir(dirname, "create_bak_dir", false);
long dir_num = 0L;
- if (!dir)
- {
- std::ostringstream oss;
- oss << "dir=\"" << dirname << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_JDIR_OPENDIR, oss.str(), "jdir", "create_bak_dir");
- }
struct dirent* entry;
while ((entry = ::readdir(dir)) != 0)
{
@@ -407,25 +376,23 @@ void
jdir::read_dir(const std::string& name, std::vector<std::string>& dir_list, const bool incl_dirs, const bool incl_files, const bool incl_links, const bool return_fqfn) {
struct stat s;
if (is_dir(name)) {
- DIR* dir = ::opendir(name.c_str());
- if (dir != 0) {
- struct dirent* entry;
- while ((entry = ::readdir(dir)) != 0) {
- if (std::strcmp(entry->d_name, ".") != 0 && std::strcmp(entry->d_name, "..") != 0) { // Ignore . and ..
- std::string full_name(name + "/" + entry->d_name);
- if (::stat(full_name.c_str(), &s))
- {
- ::closedir(dir);
- std::ostringstream oss;
- oss << "stat: file=\"" << full_name << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_JDIR_STAT, oss.str(), "jdir", "delete_dir");
- }
- if ((S_ISREG(s.st_mode) && incl_files) || (S_ISDIR(s.st_mode) && incl_dirs) || (S_ISLNK(s.st_mode) && incl_links)) {
- if (return_fqfn) {
- dir_list.push_back(name + "/" + entry->d_name);
- } else {
- dir_list.push_back(entry->d_name);
- }
+ DIR* dir = open_dir(name, "read_dir", false);
+ struct dirent* entry;
+ while ((entry = ::readdir(dir)) != 0) {
+ if (std::strcmp(entry->d_name, ".") != 0 && std::strcmp(entry->d_name, "..") != 0) { // Ignore . and ..
+ std::string full_name(name + "/" + entry->d_name);
+ if (::stat(full_name.c_str(), &s))
+ {
+ ::closedir(dir);
+ std::ostringstream oss;
+ oss << "stat: file=\"" << full_name << "\"" << FORMAT_SYSERR(errno);
+ throw jexception(jerrno::JERR_JDIR_STAT, oss.str(), "jdir", "delete_dir");
+ }
+ if ((S_ISREG(s.st_mode) && incl_files) || (S_ISDIR(s.st_mode) && incl_dirs) || (S_ISLNK(s.st_mode) && incl_links)) {
+ if (return_fqfn) {
+ dir_list.push_back(name + "/" + entry->d_name);
+ } else {
+ dir_list.push_back(entry->d_name);
}
}
}
@@ -457,6 +424,21 @@ jdir::close_dir(DIR* dir, const std::string& dir_name, const std::string& fn_nam
}
}
+DIR*
+jdir::open_dir(const std::string& dir_name, const std::string& fn_name, const bool test_enoent)
+{
+ DIR* dir = ::opendir(dir_name.c_str());
+ if (!dir) {
+ if (test_enoent && errno == ENOENT) {
+ return 0;
+ }
+ std::ostringstream oss;
+ oss << "dir=\"" << dir_name << "\"" << FORMAT_SYSERR(errno);
+ throw jexception(jerrno::JERR_JDIR_OPENDIR, oss.str(), "jdir", fn_name);
+ }
+ return dir;
+}
+
std::ostream&
operator<<(std::ostream& os, const jdir& jdir)
{
diff --git a/qpid/cpp/src/qpid/linearstore/journal/jdir.h b/qpid/cpp/src/qpid/linearstore/journal/jdir.h
index 86b16f8545..59f21ce499 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/jdir.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/jdir.h
@@ -353,6 +353,8 @@ namespace journal {
* \exception jerrno::JERR_JDIR_CLOSEDIR The directory handle could not be closed.
*/
static void close_dir(DIR* dir, const std::string& dir_name, const std::string& fn_name);
+
+ static DIR* open_dir(const std::string& dir_name, const std::string& fn_name, const bool test_enoent);
};
}}}
diff --git a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
index 6d182f40f8..abf7e58bfe 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
+++ b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
@@ -572,7 +572,7 @@ void ConnectionContext::checkClosed(boost::shared_ptr<SessionContext> ssn)
pn_session_close(ssn->session);
throw qpid::messaging::SessionError(text.str());
} else if ((pn_session_state(ssn->session) & IS_CLOSED) == IS_CLOSED) {
- throw qpid::messaging::SessionError("Session has ended");
+ throw qpid::messaging::SessionClosed();
}
}
diff --git a/qpid/cpp/src/qpid/messaging/exceptions.cpp b/qpid/cpp/src/qpid/messaging/exceptions.cpp
index 5d2683fffe..5054fdc682 100644
--- a/qpid/cpp/src/qpid/messaging/exceptions.cpp
+++ b/qpid/cpp/src/qpid/messaging/exceptions.cpp
@@ -47,6 +47,8 @@ SendError::SendError(const std::string& msg) : SenderError(msg) {}
TargetCapacityExceeded::TargetCapacityExceeded(const std::string& msg) : SendError(msg) {}
SessionError::SessionError(const std::string& msg) : MessagingException(msg) {}
+SessionClosed::SessionClosed() : SessionError("Session Closed") {}
+
TransactionError::TransactionError(const std::string& msg) : SessionError(msg) {}
TransactionAborted::TransactionAborted(const std::string& msg) : TransactionError(msg) {}
UnauthorizedAccess::UnauthorizedAccess(const std::string& msg) : SessionError(msg) {}
diff --git a/qpid/cpp/src/tests/legacystore/CMakeLists.txt b/qpid/cpp/src/tests/legacystore/CMakeLists.txt
index 85cc4be5c7..7986464eb5 100644
--- a/qpid/cpp/src/tests/legacystore/CMakeLists.txt
+++ b/qpid/cpp/src/tests/legacystore/CMakeLists.txt
@@ -36,30 +36,6 @@ if (BUILD_TESTING_UNITTESTS)
set (qpid_test_boost_libs
${Boost_UNIT_TEST_FRAMEWORK_LIBRARY} ${Boost_SYSTEM_LIBRARY})
-#
-# define_legacystore_test
-# macro to accept the name of a single source file and to create a
-# unit test executable that runs the source.
-#
-MACRO (define_legacystore_test theSourceFile)
-add_executable (legacystore_${theSourceFile}
- ${theSourceFile}
- unit_test
- ${platform_test_additions})
-target_link_libraries (legacystore_${theSourceFile}
- ${qpid_test_boost_libs}
- qpidmessaging qpidtypes qpidbroker qpidcommon legacystore_shared)
-set_target_properties (legacystore_${theSourceFile} PROPERTIES COMPILE_DEFINITIONS _IN_QPID_BROKER)
-remember_location(legacystore_${theSourceFile})
-
-add_test (legacystore_${theSourceFile} ${test_wrap} --boost-test -- ${legacystore_${theSourceFile}_LOCATION})
-ENDMACRO (define_legacystore_test)
-
-define_legacystore_test (SimpleTest)
-define_legacystore_test (OrderingTest)
-define_legacystore_test (TransactionalTest)
-define_legacystore_test (TwoPhaseCommitTest)
-
# Journal tests
MACRO (define_journal_test mainSourceFile)
if ("${ARGV1}" STREQUAL "LONG")
diff --git a/qpid/cpp/src/tests/legacystore/OrderingTest.cpp b/qpid/cpp/src/tests/legacystore/OrderingTest.cpp
deleted file mode 100644
index 74a9db1c73..0000000000
--- a/qpid/cpp/src/tests/legacystore/OrderingTest.cpp
+++ /dev/null
@@ -1,171 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "unit_test.h"
-#include "MessageUtils.h"
-
-#include "qpid/broker/Queue.h"
-#include "qpid/broker/RecoveryManagerImpl.h"
-#include "qpid/broker/PersistableObject.h"
-#include "qpid/framing/AMQHeaderBody.h"
-#include "qpid/legacystore/MessageStoreImpl.h"
-#include "qpid/log/Logger.h"
-#include "qpid/sys/Timer.h"
-
-#include <iostream>
-
-using namespace qpid;
-using namespace qpid::broker;
-using namespace qpid::framing;
-using namespace mrg::msgstore;
-
-qpid::broker::Broker::Options opts;
-qpid::broker::Broker br(opts);
-
-QPID_AUTO_TEST_SUITE(OrderingTest)
-
-#define SET_LOG_LEVEL(level) \
- qpid::log::Options opts(""); \
- opts.selectors.clear(); \
- opts.selectors.push_back(level); \
- qpid::log::Logger::instance().configure(opts);
-
-const std::string test_filename("OrderingTest");
-const char* tdp = getenv("TMP_DATA_DIR");
-const std::string test_dir(tdp && strlen(tdp) > 0 ? tdp : "/var/tmp/OrderingTest");
-
-// === Helper fns ===
-
-const std::string name("OrderingQueue");
-std::auto_ptr<MessageStoreImpl> store;
-QueueRegistry queues;
-Queue::shared_ptr queue;
-std::queue<Uuid> ids;
-
-class TestConsumer : public Consumer
-{
- public:
-
- TestConsumer(Queue::shared_ptr q, std::queue<Uuid>& i) : Consumer("test", CONSUMER), queue(q), ids(i) {};
-
- bool deliver(const QueueCursor& cursor, const Message& message)
- {
- queue->dequeue(0, cursor);
- BOOST_CHECK_EQUAL(ids.front(), MessageUtils::getMessageId(message));
- ids.pop();
- return true;
- };
- void notify() {}
- void cancel() {}
- void acknowledged(const DeliveryRecord&) {}
- OwnershipToken* getSession() { return 0; }
- private:
- Queue::shared_ptr queue;
- std::queue<Uuid>& ids;
-};
-boost::shared_ptr<TestConsumer> consumer;
-
-void setup()
-{
- store = std::auto_ptr<MessageStoreImpl>(new MessageStoreImpl(&br));
- store->init(test_dir, 4, 1, true); // truncate store
-
- queue = Queue::shared_ptr(new Queue(name, 0, store.get(), 0));
- queue->create();
- consumer = boost::shared_ptr<TestConsumer>(new TestConsumer(queue, ids));
-}
-
-void push()
-{
- Uuid messageId(true);
- ids.push(messageId);
-
- Message msg = MessageUtils::createMessage("exchange", "routing_key", messageId, true, 0);
-
- queue->deliver(msg);
-}
-
-bool pop()
-{
- return queue->dispatch(consumer);
-}
-
-void restart()
-{
- queue.reset();
- store.reset();
-
- store = std::auto_ptr<MessageStoreImpl>(new MessageStoreImpl(&br));
- store->init(test_dir, 4, 1);
- ExchangeRegistry exchanges;
- LinkRegistry links;
- sys::Timer t;
- DtxManager mgr(t);
- mgr.setStore (store.get());
- RecoveredObjects ro;
- RecoveryManagerImpl recoveryMgr(queues, exchanges, links, mgr, br.getProtocolRegistry(), ro);
- store->recover(recoveryMgr);
-
- queue = queues.find(name);
- consumer = boost::shared_ptr<TestConsumer>(new TestConsumer(queue, ids));
-}
-
-void check()
-{
- BOOST_REQUIRE(queue);
- BOOST_CHECK_EQUAL((u_int32_t) ids.size(), queue->getMessageCount());
- while (pop()) ;//keeping popping 'till all messages are dequeued
- BOOST_CHECK_EQUAL((u_int32_t) 0, queue->getMessageCount());
- BOOST_CHECK_EQUAL((size_t) 0, ids.size());
-}
-
-
-// === Test suite ===
-
-QPID_AUTO_TEST_CASE(Basic)
-{
- SET_LOG_LEVEL("error+"); // This only needs to be set once.
-
- std::cout << test_filename << ".Basic: " << std::flush;
- setup();
- //push on 10 messages
- for (int i = 0; i < 10; i++) push();
- restart();
- check();
- std::cout << "ok" << std::endl;
-}
-
-QPID_AUTO_TEST_CASE(Cycle)
-{
- std::cout << test_filename << ".Cycle: " << std::flush;
- setup();
- //push on 10 messages:
- for (int i = 0; i < 10; i++) push();
- //pop 5:
- for (int i = 0; i < 5; i++) pop();
- //push on another 5:
- for (int i = 0; i < 5; i++) push();
- restart();
- check();
- std::cout << "ok" << std::endl;
-}
-
-QPID_AUTO_TEST_SUITE_END()
diff --git a/qpid/cpp/src/tests/legacystore/SimpleTest.cpp b/qpid/cpp/src/tests/legacystore/SimpleTest.cpp
deleted file mode 100644
index d3f040817f..0000000000
--- a/qpid/cpp/src/tests/legacystore/SimpleTest.cpp
+++ /dev/null
@@ -1,500 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "unit_test.h"
-#include "MessageUtils.h"
-
-#include "qpid/broker/DirectExchange.h"
-#include "qpid/broker/Queue.h"
-#include "qpid/broker/QueueSettings.h"
-#include "qpid/broker/RecoveryManagerImpl.h"
-#include "qpid/broker/PersistableObject.h"
-#include "qpid/framing/AMQHeaderBody.h"
-#include "qpid/framing/FieldTable.h"
-#include "qpid/framing/FieldValue.h"
-#include "qpid/legacystore/MessageStoreImpl.h"
-#include "qpid/legacystore/StoreException.h"
-#include "qpid/log/Logger.h"
-#include "qpid/sys/Timer.h"
-
-#include <iostream>
-
-qpid::broker::Broker::Options opts;
-qpid::broker::Broker br(opts);
-
-#define SET_LOG_LEVEL(level) \
- qpid::log::Options opts(""); \
- opts.selectors.clear(); \
- opts.selectors.push_back(level); \
- qpid::log::Logger::instance().configure(opts);
-
-
-using boost::intrusive_ptr;
-using boost::static_pointer_cast;
-using namespace qpid;
-using namespace qpid::broker;
-using namespace qpid::framing;
-using namespace mrg::msgstore;
-using namespace std;
-
-QPID_AUTO_TEST_SUITE(SimpleTest)
-
-const string test_filename("SimpleTest");
-const char* tdp = getenv("TMP_DATA_DIR");
-const string test_dir(tdp && strlen(tdp) > 0 ? tdp : "/var/tmp/SimpleTest");
-
-// === Helper fns ===
-
-struct DummyHandler : FrameHandler
-{
- std::vector<AMQFrame> frames;
-
- virtual void handle(AMQFrame& frame){
- frames.push_back(frame);
- }
-};
-
-void recover(MessageStoreImpl& store, QueueRegistry& queues, ExchangeRegistry& exchanges, LinkRegistry& links)
-{
- sys::Timer t;
- DtxManager mgr(t);
- mgr.setStore (&store);
- RecoveredObjects ro;
- RecoveryManagerImpl recovery(queues, exchanges, links, mgr, br.getProtocolRegistry(), ro);
- store.recover(recovery);
-}
-
-void recover(MessageStoreImpl& store, ExchangeRegistry& exchanges)
-{
- QueueRegistry queues;
- LinkRegistry links;
- recover(store, queues, exchanges, links);
-}
-
-void recover(MessageStoreImpl& store, QueueRegistry& queues)
-{
- ExchangeRegistry exchanges;
- LinkRegistry links;
- recover(store, queues, exchanges, links);
-}
-
-void bindAndUnbind(const string& exchangeName, const string& queueName,
- const string& key, const FieldTable& args)
-{
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- Exchange::shared_ptr exchange(new DirectExchange(exchangeName, true, false, args));
- Queue::shared_ptr queue(new Queue(queueName, 0, &store, 0));
- store.create(*exchange, qpid::framing::FieldTable());
- store.create(*queue, qpid::framing::FieldTable());
- BOOST_REQUIRE(exchange->bind(queue, key, &args));
- store.bind(*exchange, *queue, key, args);
- }//db will be closed
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- ExchangeRegistry exchanges;
- QueueRegistry queues;
- LinkRegistry links;
-
- recover(store, queues, exchanges, links);
-
- Exchange::shared_ptr exchange = exchanges.get(exchangeName);
- Queue::shared_ptr queue = queues.find(queueName);
- // check exchange args are still set
- for (FieldTable::ValueMap::const_iterator i = args.begin(); i!=args.end(); i++) {
- BOOST_CHECK(exchange->getArgs().get((*i).first)->getData() == (*i).second->getData());
- }
- //check it is bound by unbinding
- BOOST_REQUIRE(exchange->unbind(queue, key, &args));
- store.unbind(*exchange, *queue, key, args);
- }
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- ExchangeRegistry exchanges;
- QueueRegistry queues;
- LinkRegistry links;
-
- recover(store, queues, exchanges, links);
-
- Exchange::shared_ptr exchange = exchanges.get(exchangeName);
- Queue::shared_ptr queue = queues.find(queueName);
- // check exchange args are still set
- for (FieldTable::ValueMap::const_iterator i = args.begin(); i!=args.end(); i++) {
- BOOST_CHECK(exchange->getArgs().get((*i).first)->getData() == (*i).second->getData());
- }
- //make sure it is no longer bound
- BOOST_REQUIRE(!exchange->unbind(queue, key, &args));
- }
-}
-
-
-// === Test suite ===
-
-QPID_AUTO_TEST_CASE(CreateDelete)
-{
- SET_LOG_LEVEL("error+"); // This only needs to be set once.
-
- cout << test_filename << ".CreateDelete: " << flush;
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- string name("CreateDeleteQueue");
- Queue queue(name, 0, &store, 0);
- store.create(queue, qpid::framing::FieldTable());
-// TODO - check dir exists
- BOOST_REQUIRE(queue.getPersistenceId());
- store.destroy(queue);
-// TODO - check dir is deleted
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(EmptyRecover)
-{
- cout << test_filename << ".EmptyRecover: " << flush;
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- QueueRegistry registry;
- registry.setStore (&store);
- recover(store, registry);
- //nothing to assert, just testing it doesn't blow up
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(QueueCreate)
-{
- cout << test_filename << ".QueueCreate: " << flush;
-
- uint64_t id(0);
- string name("MyDurableQueue");
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- Queue queue(name, 0, &store, 0);
- store.create(queue, qpid::framing::FieldTable());
- BOOST_REQUIRE(queue.getPersistenceId());
- id = queue.getPersistenceId();
- }//db will be closed
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- QueueRegistry registry;
- registry.setStore (&store);
- recover(store, registry);
- Queue::shared_ptr queue = registry.find(name);
- BOOST_REQUIRE(queue.get());
- BOOST_CHECK_EQUAL(id, queue->getPersistenceId());
- }
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(QueueCreateWithSettings)
-{
- cout << test_filename << ".QueueCreateWithSettings: " << flush;
-
- FieldTable arguments;
- arguments.setInt("qpid.max_count", 202);
- arguments.setInt("qpid.max_size", 1003);
- QueueSettings settings;
- settings.populate(arguments, settings.storeSettings);
- string name("MyDurableQueue");
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- Queue queue(name, settings, &store, 0);
- queue.create();
- BOOST_REQUIRE(queue.getPersistenceId());
- }//db will be closed
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- QueueRegistry registry;
- registry.setStore (&store);
- recover(store, registry);
- Queue::shared_ptr queue = registry.find(name);
- BOOST_REQUIRE(queue);
- BOOST_CHECK_EQUAL(settings.maxDepth.getCount(), 202u);
- BOOST_CHECK_EQUAL(settings.maxDepth.getSize(), 1003u);
- BOOST_CHECK_EQUAL(settings.maxDepth.getCount(), queue->getSettings().maxDepth.getCount());
- BOOST_CHECK_EQUAL(settings.maxDepth.getSize(), queue->getSettings().maxDepth.getSize());
- }
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(QueueDestroy)
-{
- cout << test_filename << ".QueueDestroy: " << flush;
-
- string name("MyDurableQueue");
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- Queue queue(name, 0, &store, 0);
- store.create(queue, qpid::framing::FieldTable());
- store.destroy(queue);
- }//db will be closed
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- QueueRegistry registry;
- registry.setStore (&store);
- recover(store, registry);
- BOOST_REQUIRE(!registry.find(name));
- }
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(Enqueue)
-{
- cout << test_filename << ".Enqueue: " << flush;
-
- //TODO: this is largely copy & paste'd from MessageTest in
- //qpid tree. ideally need some helper routines for reducing
- //this to a simpler less duplicated form
-
- string name("MyDurableQueue");
- string exchange("MyExchange");
- string routingKey("MyRoutingKey");
- Uuid messageId(true);
- string data1("abcdefg");
- string data2("hijklmn");
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- Queue::shared_ptr queue(new Queue(name, 0, &store, 0));
- queue->create();
-
- Message msg = MessageUtils::createMessage(exchange, routingKey, messageId, true, 14);
- MessageUtils::addContent(msg, data1);
- MessageUtils::addContent(msg, data2);
-
- msg.addAnnotation("abc", "xyz");
-
- queue->deliver(msg);
- }//db will be closed
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- QueueRegistry registry;
- registry.setStore (&store);
- recover(store, registry);
- Queue::shared_ptr queue = registry.find(name);
- BOOST_REQUIRE(queue);
- BOOST_CHECK_EQUAL((u_int32_t) 1, queue->getMessageCount());
- Message msg = MessageUtils::get(*queue);
-
- BOOST_CHECK_EQUAL(routingKey, msg.getRoutingKey());
- BOOST_CHECK_EQUAL(messageId, MessageUtils::getMessageId(msg));
- BOOST_CHECK_EQUAL(std::string("xyz"), msg.getAnnotation("abc"));
- BOOST_CHECK_EQUAL((u_int64_t) 14, msg.getContent().size());
-
- DummyHandler handler;
- MessageUtils::deliver(msg, handler, 100);
- BOOST_CHECK_EQUAL((size_t) 2, handler.frames.size());
- AMQContentBody* contentBody(dynamic_cast<AMQContentBody*>(handler.frames[1].getBody()));
- BOOST_REQUIRE(contentBody);
- BOOST_CHECK_EQUAL(data1.size() + data2.size(), contentBody->getData().size());
- BOOST_CHECK_EQUAL(data1 + data2, contentBody->getData());
- }
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(Dequeue)
-{
- cout << test_filename << ".Dequeue: " << flush;
-
- //TODO: reduce the duplication in these tests
- string name("MyDurableQueue");
- {
- string exchange("MyExchange");
- string routingKey("MyRoutingKey");
- Uuid messageId(true);
- string data("abcdefg");
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- Queue::shared_ptr queue(new Queue(name, 0, &store, 0));
- queue->create();
-
- Message msg = MessageUtils::createMessage(exchange, routingKey, messageId, true, 7);
- MessageUtils::addContent(msg, data);
-
- queue->deliver(msg);
-
- QueueCursor cursor;
- MessageUtils::get(*queue, &cursor);
- queue->dequeue(0, cursor);
- }//db will be closed
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- QueueRegistry registry;
- registry.setStore (&store);
- recover(store, registry);
- Queue::shared_ptr queue = registry.find(name);
- BOOST_REQUIRE(queue);
- BOOST_CHECK_EQUAL((u_int32_t) 0, queue->getMessageCount());
- }
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(ExchangeCreateAndDestroy)
-{
- cout << test_filename << ".ExchangeCreateAndDestroy: " << flush;
-
- uint64_t id(0);
- string name("MyDurableExchange");
- string type("direct");
- FieldTable args;
- args.setString("a", "A");
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- ExchangeRegistry registry;
- Exchange::shared_ptr exchange = registry.declare(name, type, true, false, args).first;
- store.create(*exchange, qpid::framing::FieldTable());
- id = exchange->getPersistenceId();
- BOOST_REQUIRE(id);
- }//db will be closed
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- ExchangeRegistry registry;
-
- recover(store, registry);
-
- Exchange::shared_ptr exchange = registry.get(name);
- BOOST_CHECK_EQUAL(id, exchange->getPersistenceId());
- BOOST_CHECK_EQUAL(type, exchange->getType());
- BOOST_REQUIRE(exchange->isDurable());
- BOOST_CHECK_EQUAL(*args.get("a"), *exchange->getArgs().get("a"));
- store.destroy(*exchange);
- }
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- ExchangeRegistry registry;
-
- recover(store, registry);
-
- try {
- Exchange::shared_ptr exchange = registry.get(name);
- BOOST_FAIL("Expected exchange not to be found");
- } catch (const SessionException& e) {
- BOOST_CHECK_EQUAL((framing::ReplyCode) 404, e.code);
- }
- }
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(ExchangeBindAndUnbind)
-{
- cout << test_filename << ".ExchangeBindAndUnbind: " << flush;
-
- bindAndUnbind("MyDurableExchange", "MyDurableQueue", "my-routing-key", FieldTable());
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(ExchangeBindAndUnbindWithArgs)
-{
- cout << test_filename << ".ExchangeBindAndUnbindWithArgs: " << flush;
-
- FieldTable args;
- args.setString("a", "A");
- args.setString("b", "B");
- bindAndUnbind("MyDurableExchange", "MyDurableQueue", "my-routing-key", args);
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(ExchangeImplicitUnbind)
-{
- cout << test_filename << ".ExchangeImplicitUnbind: " << flush;
-
- string exchangeName("MyDurableExchange");
- string queueName1("MyDurableQueue1");
- string queueName2("MyDurableQueue2");
- string key("my-routing-key");
- FieldTable args;
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1, true); // truncate store
- Exchange::shared_ptr exchange(new DirectExchange(exchangeName, true, false, args));
- Queue::shared_ptr queue1(new Queue(queueName1, 0, &store, 0));
- Queue::shared_ptr queue2(new Queue(queueName2, 0, &store, 0));
- store.create(*exchange, qpid::framing::FieldTable());
- store.create(*queue1, qpid::framing::FieldTable());
- store.create(*queue2, qpid::framing::FieldTable());
- store.bind(*exchange, *queue1, key, args);
- store.bind(*exchange, *queue2, key, args);
- //delete queue1:
- store.destroy(*queue1);
- }//db will be closed
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- ExchangeRegistry exchanges;
- QueueRegistry queues;
- LinkRegistry links;
-
- //ensure recovery works ok:
- recover(store, queues, exchanges, links);
-
- Exchange::shared_ptr exchange = exchanges.get(exchangeName);
- BOOST_REQUIRE(!queues.find(queueName1).get());
- BOOST_REQUIRE(queues.find(queueName2).get());
-
- //delete exchange:
- store.destroy(*exchange);
- }
- {
- MessageStoreImpl store(&br);
- store.init(test_dir, 4, 1);
- ExchangeRegistry exchanges;
- QueueRegistry queues;
- LinkRegistry links;
-
- //ensure recovery works ok:
- recover(store, queues, exchanges, links);
-
- try {
- Exchange::shared_ptr exchange = exchanges.get(exchangeName);
- BOOST_FAIL("Expected exchange not to be found");
- } catch (const SessionException& e) {
- BOOST_CHECK_EQUAL((framing::ReplyCode) 404, e.code);
- }
- Queue::shared_ptr queue = queues.find(queueName2);
- store.destroy(*queue);
- }
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_SUITE_END()
diff --git a/qpid/cpp/src/tests/legacystore/TransactionalTest.cpp b/qpid/cpp/src/tests/legacystore/TransactionalTest.cpp
deleted file mode 100644
index d1bc34d5a7..0000000000
--- a/qpid/cpp/src/tests/legacystore/TransactionalTest.cpp
+++ /dev/null
@@ -1,354 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "unit_test.h"
-#include "MessageUtils.h"
-
-#include "qpid/broker/Queue.h"
-#include "qpid/broker/RecoveryManagerImpl.h"
-#include "qpid/broker/PersistableObject.h"
-#include "qpid/framing/AMQHeaderBody.h"
-#include "qpid/legacystore/MessageStoreImpl.h"
-#include "qpid/legacystore/StoreException.h"
-#include "qpid/log/Statement.h"
-#include "qpid/log/Logger.h"
-#include "qpid/sys/Timer.h"
-
-#include <iostream>
-
-using namespace mrg::msgstore;
-using namespace qpid;
-using namespace qpid::broker;
-using namespace qpid::framing;
-using namespace std;
-
-namespace {
-qpid::broker::Broker::Options opts;
-qpid::broker::Broker br(opts);
-}
-
-QPID_AUTO_TEST_SUITE(TransactionalTest)
-
-#define SET_LOG_LEVEL(level) \
- qpid::log::Options opts(""); \
- opts.selectors.clear(); \
- opts.selectors.push_back(level); \
- qpid::log::Logger::instance().configure(opts);
-
-const string test_filename("TransactionalTest");
-const char* tdp = getenv("TMP_DATA_DIR");
-const string test_dir(tdp && strlen(tdp) > 0 ? tdp : "/var/tmp/TransactionalTest");
-
-// Test txn context which has special setCompleteFailure() method which prevents entire "txn complete" process from hapenning
-class TestTxnCtxt : public TxnCtxt
-{
- public:
- TestTxnCtxt(IdSequence* _loggedtx) : TxnCtxt(_loggedtx) {}
- void setCompleteFailure(const unsigned num_queues_rem) {
- // Remove queue members from back of impactedQueues until queues_rem reamin.
- // to end to simulate multi-queue txn complete failure.
- while (impactedQueues.size() > num_queues_rem) impactedQueues.erase(impactedQueues.begin());
- }
- void resetPreparedXidStorePtr() { preparedXidStorePtr = 0; }
-};
-
-// Test store which has special begin() which returns a TestTPCTxnCtxt, and a method to check for
-// remaining open transactions.
-// begin(), commit(), and abort() all hide functions in MessageStoreImpl. To avoid the compiler
-// warnings/errors these are renamed with a 'TMS' prefix.
-class TestMessageStore: public MessageStoreImpl
-{
- public:
- TestMessageStore(qpid::broker::Broker* br, const char* envpath = 0) : MessageStoreImpl(br, envpath) {}
- std::auto_ptr<qpid::broker::TransactionContext> TMSbegin() {
- checkInit();
- // pass sequence number for c/a
- return auto_ptr<TransactionContext>(new TestTxnCtxt(&messageIdSequence));
- }
- void TMScommit(TransactionContext& ctxt, const bool complete_prepared_list) {
- checkInit();
- TxnCtxt* txn(check(&ctxt));
- if (!txn->isTPC()) {
- localPrepare(dynamic_cast<TxnCtxt*>(txn));
- if (!complete_prepared_list) dynamic_cast<TestTxnCtxt*>(txn)->resetPreparedXidStorePtr();
- }
- completed(*dynamic_cast<TxnCtxt*>(txn), true);
- }
- void TMSabort(TransactionContext& ctxt, const bool complete_prepared_list)
- {
- checkInit();
- TxnCtxt* txn(check(&ctxt));
- if (!txn->isTPC()) {
- localPrepare(dynamic_cast<TxnCtxt*>(txn));
- if (!complete_prepared_list) dynamic_cast<TestTxnCtxt*>(txn)->resetPreparedXidStorePtr();
- }
- completed(*dynamic_cast<TxnCtxt*>(txn), false);
- }
-};
-
-// === Helper fns ===
-
-const string nameA("queueA");
-const string nameB("queueB");
-//const Uuid messageId(true);
-std::auto_ptr<MessageStoreImpl> store;
-std::auto_ptr<QueueRegistry> queues;
-Queue::shared_ptr queueA;
-Queue::shared_ptr queueB;
-
-template <class T>
-void setup()
-{
- store = std::auto_ptr<T>(new T(&br));
- store->init(test_dir, 4, 1, true); // truncate store
-
- //create two queues:
- queueA = Queue::shared_ptr(new Queue(nameA, 0, store.get(), 0));
- queueA->create();
- queueB = Queue::shared_ptr(new Queue(nameB, 0, store.get(), 0));
- queueB->create();
-}
-
-template <class T>
-void restart()
-{
- queueA.reset();
- queueB.reset();
- queues.reset();
- store.reset();
-
- store = std::auto_ptr<T>(new T(&br));
- store->init(test_dir, 4, 1);
- queues = std::auto_ptr<QueueRegistry>(new QueueRegistry);
- ExchangeRegistry exchanges;
- LinkRegistry links;
- sys::Timer t;
- DtxManager mgr(t);
- mgr.setStore (store.get());
- RecoveredObjects ro;
- RecoveryManagerImpl recovery(*queues, exchanges, links, mgr, br.getProtocolRegistry(), ro);
- store->recover(recovery);
-
- queueA = queues->find(nameA);
- queueB = queues->find(nameB);
-}
-
-Message createMessage(const string& id, const string& exchange="exchange", const string& key="routing_key")
-{
- return MessageUtils::createMessage(exchange, key, Uuid(), true, 0, id);
-}
-
-void checkMsg(Queue::shared_ptr& queue, u_int32_t size, const string& msgid = "<none>")
-{
- BOOST_REQUIRE(queue);
- BOOST_CHECK_EQUAL(size, queue->getMessageCount());
- if (size > 0) {
- Message msg = MessageUtils::get(*queue);
- BOOST_REQUIRE(msg);
- BOOST_CHECK_EQUAL(msgid, MessageUtils::getCorrelationId(msg));
- }
-}
-
-void swap(bool commit)
-{
- setup<MessageStoreImpl>();
-
- //create message and enqueue it onto first queue:
- Message msgA = createMessage("Message", "exchange", "routing_key");
- queueA->deliver(msgA);
-
- QueueCursor cursorB;
- Message msgB = MessageUtils::get(*queueA, &cursorB);
- BOOST_REQUIRE(msgB);
- //move the message from one queue to the other as a transaction
- std::auto_ptr<TransactionContext> txn = store->begin();
- TxBuffer tx;
- queueB->deliver(msgB, &tx);//note: need to enqueue it first to avoid message being deleted
-
- queueA->dequeue(txn.get(), cursorB);
- tx.prepare(txn.get());
- if (commit) {
- store->commit(*txn);
- } else {
- store->abort(*txn);
- }
-
- restart<MessageStoreImpl>();
-
- // Check outcome
- BOOST_REQUIRE(queueA);
- BOOST_REQUIRE(queueB);
-
- Queue::shared_ptr x;//the queue from which the message was swapped
- Queue::shared_ptr y;//the queue on which the message is expected to be
-
- if (commit) {
- x = queueA;
- y = queueB;
- } else {
- x = queueB;
- y = queueA;
- }
-
- checkMsg(x, 0);
- checkMsg(y, 1, "Message");
- checkMsg(y, 0);
-}
-
-void testMultiQueueTxn(const unsigned num_queues_rem, const bool complete_prepared_list, const bool commit)
-{
- setup<TestMessageStore>();
- TestMessageStore* tmsp = static_cast<TestMessageStore*>(store.get());
- std::auto_ptr<TransactionContext> txn(tmsp->TMSbegin());
- TxBuffer tx;
-
- //create two messages and enqueue them onto both queues:
- Message msgA = createMessage("MessageA", "exchange", "routing_key");
- queueA->deliver(msgA, &tx);
- queueB->deliver(msgA, &tx);
- Message msgB = createMessage("MessageB", "exchange", "routing_key");
- queueA->deliver(msgB, &tx);
- queueB->deliver(msgB, &tx);
-
- tx.prepare(txn.get());
- static_cast<TestTxnCtxt*>(txn.get())->setCompleteFailure(num_queues_rem);
- if (commit)
- tmsp->TMScommit(*txn, complete_prepared_list);
- else
- tmsp->TMSabort(*txn, complete_prepared_list);
- restart<TestMessageStore>();
-
- // Check outcome
- if (commit)
- {
- checkMsg(queueA, 2, "MessageA");
- checkMsg(queueB, 2, "MessageA");
- checkMsg(queueA, 1, "MessageB");
- checkMsg(queueB, 1, "MessageB");
- }
- checkMsg(queueA, 0);
- checkMsg(queueB, 0);
-}
-
-// === Test suite ===
-
-QPID_AUTO_TEST_CASE(Commit)
-{
- SET_LOG_LEVEL("error+"); // This only needs to be set once.
-
- cout << test_filename << ".Commit: " << flush;
- swap(true);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(Abort)
-{
- cout << test_filename << ".Abort: " << flush;
- swap(false);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueCommit)
-{
- cout << test_filename << ".MultiQueueCommit: " << flush;
- testMultiQueueTxn(2, true, true);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueAbort)
-{
- cout << test_filename << ".MultiQueueAbort: " << flush;
- testMultiQueueTxn(2, true, false);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueNoQueueCommitRecover)
-{
- cout << test_filename << ".MultiQueueNoQueueCommitRecover: " << flush;
- testMultiQueueTxn(0, false, true);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueNoQueueAbortRecover)
-{
- cout << test_filename << ".MultiQueueNoQueueAbortRecover: " << flush;
- testMultiQueueTxn(0, false, false);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueSomeQueueCommitRecover)
-{
- cout << test_filename << ".MultiQueueSomeQueueCommitRecover: " << flush;
- testMultiQueueTxn(1, false, true);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueSomeQueueAbortRecover)
-{
- cout << test_filename << ".MultiQueueSomeQueueAbortRecover: " << flush;
- testMultiQueueTxn(1, false, false);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueAllQueueCommitRecover)
-{
- cout << test_filename << ".MultiQueueAllQueueCommitRecover: " << flush;
- testMultiQueueTxn(2, false, true);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueAllQueueAbortRecover)
-{
- cout << test_filename << ".MultiQueueAllQueueAbortRecover: " << flush;
- testMultiQueueTxn(2, false, false);
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(LockedRecordTest)
-{
- cout << test_filename << ".LockedRecordTest: " << flush;
-
- setup<MessageStoreImpl>();
- queueA->deliver(createMessage("Message", "exchange", "routingKey"));
- std::auto_ptr<TransactionContext> txn = store->begin();
-
- QueueCursor cursor;
- Message msg = MessageUtils::get(*queueA, &cursor);
- queueA->dequeue(txn.get(), cursor);
-
- try {
- store->dequeue(0, msg.getPersistentContext(), *queueA);
- BOOST_ERROR("Did not throw JERR_MAP_LOCKED exception as expected.");
- }
- catch (const mrg::msgstore::StoreException& e) {
- if (std::strstr(e.what(), "JERR_MAP_LOCKED") == 0)
- BOOST_ERROR("Unexpected StoreException: " << e.what());
- }
- catch (const std::exception& e) {
- BOOST_ERROR("Unexpected exception: " << e.what());
- }
- store->commit(*txn);
- checkMsg(queueA, 0);
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_SUITE_END()
diff --git a/qpid/cpp/src/tests/legacystore/TwoPhaseCommitTest.cpp b/qpid/cpp/src/tests/legacystore/TwoPhaseCommitTest.cpp
deleted file mode 100644
index 25bb9dc607..0000000000
--- a/qpid/cpp/src/tests/legacystore/TwoPhaseCommitTest.cpp
+++ /dev/null
@@ -1,678 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "unit_test.h"
-#include "MessageUtils.h"
-
-#include "qpid/broker/Queue.h"
-#include "qpid/broker/RecoveryManagerImpl.h"
-#include "qpid/broker/PersistableObject.h"
-#include "qpid/framing/AMQHeaderBody.h"
-#include "qpid/legacystore/MessageStoreImpl.h"
-#include "qpid/legacystore/TxnCtxt.h"
-#include "qpid/log/Logger.h"
-#include "qpid/log/Statement.h"
-#include "qpid/sys/Timer.h"
-
-#include <iostream>
-
-using namespace mrg::msgstore;
-using namespace qpid;
-using namespace qpid::broker;
-using namespace qpid::framing;
-using namespace std;
-
-
-qpid::broker::Broker::Options opts;
-qpid::broker::Broker br(opts);
-
-
-QPID_AUTO_TEST_SUITE(TwoPhaseCommitTest)
-
-#define SET_LOG_LEVEL(level) \
- qpid::log::Options opts(""); \
- opts.selectors.clear(); \
- opts.selectors.push_back(level); \
- qpid::log::Logger::instance().configure(opts);
-
-
-const string test_filename("TwoPhaseCommitTest");
-const char* tdp = getenv("TMP_DATA_DIR");
-string test_dir(tdp && strlen(tdp) > 0 ? tdp : "/var/tmp/TwoPhaseCommitTest");
-
-// === Helper fns ===
-
-class TwoPhaseCommitTest
-{
-
- class Strategy
- {
- public:
- virtual void init() = 0;
- virtual void run(TPCTransactionContext* txn) = 0;
- virtual void check(bool committed) = 0;
- virtual ~Strategy(){}
- };
-
- class Swap : public Strategy
- {
- TwoPhaseCommitTest* const test;
- const string messageId;
- Message msg;
- public:
- Swap(TwoPhaseCommitTest* const test_, const string& messageId_): test(test_), messageId(messageId_) {}
- void init(){ msg = test->deliver(messageId, test->queueA); }
- void run(TPCTransactionContext* txn) { test->swap(txn, test->queueA, test->queueB); }
- void check(bool committed) { test->swapCheck(committed, messageId, test->queueA, test->queueB); }
- };
-
- class Enqueue : public Strategy
- {
- TwoPhaseCommitTest* const test;
- Message msg1;
- Message msg2;
- Message msg3;
- public:
- Enqueue(TwoPhaseCommitTest* const test_): test(test_) {}
- void init() {}
- void run(TPCTransactionContext* txn) {
- msg1 = test->enqueue(txn, "Enqueue1", test->queueA);
- msg2 = test->enqueue(txn, "Enqueue2", test->queueA);
- msg3 = test->enqueue(txn, "Enqueue3", test->queueA);
- }
- void check(bool committed) {
- if (committed) {
- test->checkMsg(test->queueA, 3, "Enqueue1");
- test->checkMsg(test->queueA, 2, "Enqueue2");
- test->checkMsg(test->queueA, 1, "Enqueue3");
- }
- test->checkMsg(test->queueA, 0);
- }
- };
-
- class Dequeue : public Strategy
- {
- TwoPhaseCommitTest* const test;
- Message msg1;
- Message msg2;
- Message msg3;
- public:
- Dequeue(TwoPhaseCommitTest* const test_): test(test_) {}
- void init() {
- msg1 = test->deliver("Dequeue1", test->queueA);
- msg2 = test->deliver("Dequeue2", test->queueA);
- msg3 = test->deliver("Dequeue3", test->queueA);
- }
- void run(TPCTransactionContext* txn) {
- test->dequeue(txn, test->queueA);
- test->dequeue(txn, test->queueA);
- test->dequeue(txn, test->queueA);
- }
- void check(bool committed) {
- if (!committed) {
- test->checkMsg(test->queueA, 3, "Dequeue1");
- test->checkMsg(test->queueA, 2, "Dequeue2");
- test->checkMsg(test->queueA, 1, "Dequeue3");
- }
- test->checkMsg(test->queueA, 0);
- }
- };
-
- class MultiQueueTxn : public Strategy
- {
- TwoPhaseCommitTest* const test;
- Message msg1;
- Message msg2;
- std::set<Queue::shared_ptr> queueset;
- public:
- MultiQueueTxn(TwoPhaseCommitTest* const test_): test(test_) {}
- virtual void init() {}
- virtual void run(TPCTransactionContext* txn) {
- queueset.insert(test->queueA);
- queueset.insert(test->queueB);
- msg1 = test->enqueue(txn, "Message1", queueset);
- msg2 = test->enqueue(txn, "Message2", queueset);
- queueset.clear();
- }
- virtual void check(bool committed) {
- TestMessageStore* sptr = static_cast<TestMessageStore*>(test->store.get());
- if (committed)
- {
- test->checkMsg(test->queueA, 2, "Message1");
- test->checkMsg(test->queueB, 2, "Message1");
- test->checkMsg(test->queueA, 1, "Message2");
- test->checkMsg(test->queueB, 1, "Message2");
- }
- test->checkMsg(test->queueA, 0);
- test->checkMsg(test->queueB, 0);
- // Check there are no remaining open txns in store
- BOOST_CHECK_EQUAL(u_int32_t(0), sptr->getRemainingTxns(*(test->queueA)));
- BOOST_CHECK_EQUAL(u_int32_t(0), sptr->getRemainingTxns(*(test->queueB)));
- BOOST_CHECK_EQUAL(u_int32_t(0), sptr->getRemainingPreparedListTxns());
- }
- };
-
- // Test txn context which has special setCompleteFailure() method which prevents entire "txn complete" process from hapenning
- class TestTPCTxnCtxt : public TPCTxnCtxt
- {
- public:
- TestTPCTxnCtxt(const std::string& _xid, IdSequence* _loggedtx) : TPCTxnCtxt(_xid, _loggedtx) {}
- void setCompleteFailure(const unsigned num_queues_rem, const bool complete_prepared_list) {
- // Remove queue members from back of impactedQueues until queues_rem reamin.
- // to end to simulate multi-queue txn complete failure.
- while (impactedQueues.size() > num_queues_rem) impactedQueues.erase(impactedQueues.begin());
- // If prepared list is not to be committed, set pointer to 0
- if (!complete_prepared_list) preparedXidStorePtr = 0;
- }
- };
-
- // Test store which has sepcial begin() which returns a TestTPCTxnCtxt, and a method to check for
- // reamining open transactions
- class TestMessageStore: public MessageStoreImpl
- {
- public:
- TestMessageStore(qpid::broker::Broker* br, const char* envpath = 0) : MessageStoreImpl(br, envpath) {}
- std::auto_ptr<qpid::broker::TPCTransactionContext> TMSbegin(const std::string& xid) {
- checkInit();
- IdSequence* jtx = &messageIdSequence;
- // pass sequence number for c/a
- return auto_ptr<TPCTransactionContext>(new TestTPCTxnCtxt(xid, jtx));
- }
- u_int32_t getRemainingTxns(const PersistableQueue& queue) {
- return static_cast<JournalImpl*>(queue.getExternalQueueStore())->get_open_txn_cnt();
- }
- u_int32_t getRemainingPreparedListTxns() {
- return tplStorePtr->get_open_txn_cnt();
- }
- };
-
- const string nameA;
- const string nameB;
- std::auto_ptr<MessageStoreImpl> store;
- std::auto_ptr<DtxManager> dtxmgr;
- std::auto_ptr<QueueRegistry> queues;
- std::auto_ptr<LinkRegistry> links;
- Queue::shared_ptr queueA;
- Queue::shared_ptr queueB;
- Message msg1;
- Message msg2;
- Message msg4;
- std::auto_ptr<TxBuffer> tx;
-
- void recoverPrepared(bool commit)
- {
- setup<MessageStoreImpl>();
-
- Swap swap(this, "RecoverPrepared");
- swap.init();
- std::auto_ptr<TPCTransactionContext> txn(store->begin("my-xid"));
- swap.run(txn.get());
- if (tx.get()) {
- tx->prepare(txn.get());
- tx.reset();
- }
-
- store->prepare(*txn);
- restart<MessageStoreImpl>();
-
- //check that the message is not available from either queue
- BOOST_CHECK_EQUAL((u_int32_t) 0, queueA->getMessageCount());
- BOOST_CHECK_EQUAL((u_int32_t) 0, queueB->getMessageCount());
-
- //commit/abort the txn - through the dtx manager, not directly on the store
- if (commit) {
- dtxmgr->commit("my-xid", false);
- } else {
- dtxmgr->rollback("my-xid");
- }
-
- swap.check(commit);
- restart<MessageStoreImpl>();
- swap.check(commit);
- }
-
- void testMultiQueueTxn(const unsigned num_queues_rem, const bool complete_prepared_list, const bool commit)
- {
- setup<TestMessageStore>();
- MultiQueueTxn mqtTest(this);
- mqtTest.init();
- std::auto_ptr<TPCTransactionContext> txn(static_cast<TestMessageStore*>(store.get())->begin("my-xid"));
- mqtTest.run(txn.get());
- if (tx.get()) {
- tx->prepare(txn.get());
- tx.reset();
- }
- store->prepare(*txn);
-
- // As the commits and aborts should happen through DtxManager, and it is too complex to
- // pass all these test params through, we bypass DtxManager and use the store directly.
- // This will prevent the queues from seeing committed txns, however. To test the success
- // or failure of
- static_cast<TestTPCTxnCtxt*>(txn.get())->setCompleteFailure(num_queues_rem, complete_prepared_list);
- if (commit)
- store->commit(*txn);
- else
- store->abort(*txn);
- restart<TestMessageStore>();
- mqtTest.check(commit);
- }
-
- void commit(Strategy& strategy)
- {
- setup<MessageStoreImpl>();
- strategy.init();
-
- std::auto_ptr<TPCTransactionContext> txn(store->begin("my-xid"));
- strategy.run(txn.get());
- if (tx.get()) {
- tx->prepare(txn.get());
- tx.reset();
- }
- store->prepare(*txn);
- store->commit(*txn);
- restart<MessageStoreImpl>();
- strategy.check(true);
- }
-
- void abort(Strategy& strategy, bool prepare)
- {
- setup<MessageStoreImpl>();
- strategy.init();
-
- std::auto_ptr<TPCTransactionContext> txn(store->begin("my-xid"));
- strategy.run(txn.get());
- if (tx.get()) {
- tx->prepare(txn.get());
- tx.reset();
- }
- if (prepare) store->prepare(*txn);
- store->abort(*txn);
- restart<MessageStoreImpl>();
- strategy.check(false);
- }
-
- void swap(TPCTransactionContext* txn, Queue::shared_ptr& from, Queue::shared_ptr& to)
- {
- QueueCursor c;
- Message msg1 = MessageUtils::get(*from, &c);//just dequeues in memory
- //move the message from one queue to the other as part of a
- //distributed transaction
- if (!tx.get()) tx = std::auto_ptr<TxBuffer>(new TxBuffer);
- to->deliver(msg1, tx.get());//note: need to enqueue it first to avoid message being deleted
- from->dequeue(txn, c);
- }
-
- void dequeue(TPCTransactionContext* txn, Queue::shared_ptr& queue)
- {
- QueueCursor c;
- Message msg2 = MessageUtils::get(*queue, &c);//just dequeues in memory
- queue->dequeue(txn, c);
- }
-
- Message enqueue(TPCTransactionContext* /*txn*/, const string& msgid, Queue::shared_ptr& queue)
- {
- Message msg = createMessage(msgid);
- if (!tx.get()) tx = std::auto_ptr<TxBuffer>(new TxBuffer);
- queue->deliver(msg, tx.get());
- return msg;
- }
-
- Message enqueue(TPCTransactionContext* /*txn*/, const string& msgid, std::set<Queue::shared_ptr>& queueset)
- {
- if (!tx.get()) tx = std::auto_ptr<TxBuffer>(new TxBuffer);
- Message msg = createMessage(msgid);
- for (std::set<Queue::shared_ptr>::iterator i = queueset.begin(); i != queueset.end(); i++) {
- (*i)->deliver(msg, tx.get());
- }
- return msg;
- }
-
- Message deliver(const string& msgid, Queue::shared_ptr& queue)
- {
- Message m = createMessage(msgid);
- queue->deliver(m);
- return m;
- }
-
- template <class T>
- void setup()
- {
- store = std::auto_ptr<T>(new T(&br));
- store->init(test_dir, 4, 1, true); // truncate store
-
- //create two queues:
- queueA = Queue::shared_ptr(new Queue(nameA, 0, store.get(), 0));
- queueA->create();
- queueB = Queue::shared_ptr(new Queue(nameB, 0, store.get(), 0));
- queueB->create();
- }
-
- Message createMessage(const string& id, const string& exchange="exchange", const string& key="routing_key")
- {
- Message msg = MessageUtils::createMessage(exchange, key, Uuid(), true, 0, id);
- return msg;
- }
-
- template <class T>
- void restart()
- {
- queueA.reset();
- queueB.reset();
- store.reset();
- queues.reset();
- links.reset();
-
- store = std::auto_ptr<T>(new T(&br));
- store->init(test_dir, 4, 1);
- sys::Timer t;
- ExchangeRegistry exchanges;
- queues = std::auto_ptr<QueueRegistry>(new QueueRegistry);
- links = std::auto_ptr<LinkRegistry>(new LinkRegistry);
- dtxmgr = std::auto_ptr<DtxManager>(new DtxManager(t));
- dtxmgr->setStore (store.get());
- RecoveredObjects ro;
- RecoveryManagerImpl recovery(*queues, exchanges, *links, *dtxmgr, br.getProtocolRegistry(), ro);
- store->recover(recovery);
-
- queueA = queues->find(nameA);
- queueB = queues->find(nameB);
- }
-
- void checkMsg(Queue::shared_ptr& queue, u_int32_t size, const string& msgid = "<none>")
- {
- BOOST_REQUIRE(queue);
- BOOST_CHECK_EQUAL(size, queue->getMessageCount());
- if (size > 0) {
- Message msg = MessageUtils::get(*queue);
- BOOST_REQUIRE(msg);
- BOOST_CHECK_EQUAL(msgid, MessageUtils::getCorrelationId(msg));
- }
- }
-
- void swapCheck(bool swapped, const string& msgid, Queue::shared_ptr& from, Queue::shared_ptr& to)
- {
- BOOST_REQUIRE(from);
- BOOST_REQUIRE(to);
-
- Queue::shared_ptr x; //the queue from which the message was swapped
- Queue::shared_ptr y; //the queue on which the message is expected to be
-
- if (swapped) {
- x = from;
- y = to;
- } else {
- x = to;
- y = from;
- }
-
- checkMsg(x, 0);
- checkMsg(y, 1, msgid);
- checkMsg(y, 0);
- }
-
-public:
- TwoPhaseCommitTest() : nameA("queueA"), nameB("queueB") {}
-
- void testCommitEnqueue()
- {
- Enqueue enqueue(this);
- commit(enqueue);
- }
-
- void testCommitDequeue()
- {
- Dequeue dequeue(this);
- commit(dequeue);
- }
-
- void testCommitSwap()
- {
- Swap swap(this, "SwapMessageId");
- commit(swap);
- }
-
- void testPrepareAndAbortEnqueue()
- {
- Enqueue enqueue(this);
- abort(enqueue, true);
- }
-
- void testPrepareAndAbortDequeue()
- {
- Dequeue dequeue(this);
- abort(dequeue, true);
- }
-
- void testPrepareAndAbortSwap()
- {
- Swap swap(this, "SwapMessageId");
- abort(swap, true);
- }
-
- void testAbortNoPrepareEnqueue()
- {
- Enqueue enqueue(this);
- abort(enqueue, false);
- }
-
- void testAbortNoPrepareDequeue()
- {
- Dequeue dequeue(this);
- abort(dequeue, false);
- }
-
- void testAbortNoPrepareSwap()
- {
- Swap swap(this, "SwapMessageId");
- abort(swap, false);
- }
-
- void testRecoverPreparedThenCommitted()
- {
- recoverPrepared(true);
- }
-
- void testRecoverPreparedThenAborted()
- {
- recoverPrepared(false);
- }
-
- void testMultiQueueCommit()
- {
- testMultiQueueTxn(2, true, true);
- }
-
- void testMultiQueueAbort()
- {
- testMultiQueueTxn(2, true, false);
- }
-
- void testMultiQueueNoQueueCommitRecover()
- {
- testMultiQueueTxn(0, false, true);
- }
-
- void testMultiQueueNoQueueAbortRecover()
- {
- testMultiQueueTxn(0, false, false);
- }
-
- void testMultiQueueSomeQueueCommitRecover()
- {
- testMultiQueueTxn(1, false, true);
- }
-
- void testMultiQueueSomeQueueAbortRecover()
- {
- testMultiQueueTxn(1, false, false);
- }
-
- void testMultiQueueAllQueueCommitRecover()
- {
- testMultiQueueTxn(2, false, true);
- }
-
- void testMultiQueueAllQueueAbortRecover()
- {
- testMultiQueueTxn(2, false, false);
- }
-};
-
-TwoPhaseCommitTest tpct;
-
-// === Test suite ===
-
-QPID_AUTO_TEST_CASE(CommitEnqueue)
-{
- SET_LOG_LEVEL("error+"); // This only needs to be set once.
-
- cout << test_filename << ".CommitEnqueue: " << flush;
- tpct.testCommitEnqueue();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(CommitDequeue)
-{
- cout << test_filename << ".CommitDequeue: " << flush;
- tpct.testCommitDequeue();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(CommitSwap)
-{
- cout << test_filename << ".CommitSwap: " << flush;
- tpct.testCommitSwap();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(PrepareAndAbortEnqueue)
-{
- cout << test_filename << ".PrepareAndAbortEnqueue: " << flush;
- tpct.testPrepareAndAbortEnqueue();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(PrepareAndAbortDequeue)
-{
- cout << test_filename << ".PrepareAndAbortDequeue: " << flush;
- tpct.testPrepareAndAbortDequeue();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(PrepareAndAbortSwap)
-{
- cout << test_filename << ".PrepareAndAbortSwap: " << flush;
- tpct.testPrepareAndAbortSwap();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(AbortNoPrepareEnqueue)
-{
- cout << test_filename << ".AbortNoPrepareEnqueue: " << flush;
- tpct.testAbortNoPrepareEnqueue();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(AbortNoPrepareDequeue)
-{
- cout << test_filename << ".AbortNoPrepareDequeue: " << flush;
- tpct.testAbortNoPrepareDequeue();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(AbortNoPrepareSwap)
-{
- cout << test_filename << ".AbortNoPrepareSwap: " << flush;
- tpct.testAbortNoPrepareSwap();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(RecoverPreparedThenCommitted)
-{
- cout << test_filename << ".RecoverPreparedThenCommitted: " << flush;
- tpct.testRecoverPreparedThenCommitted();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(RecoverPreparedThenAborted)
-{
- cout << test_filename << ".RecoverPreparedThenAborted: " << flush;
- tpct.testRecoverPreparedThenAborted();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueCommit)
-{
- cout << test_filename << ".MultiQueueCommit: " << flush;
- tpct.testMultiQueueCommit();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueAbort)
-{
- cout << test_filename << ".MultiQueueAbort: " << flush;
- tpct.testMultiQueueAbort();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueNoQueueCommitRecover)
-{
- cout << test_filename << ".MultiQueueNoQueueCommitRecover: " << flush;
- tpct.testMultiQueueNoQueueCommitRecover();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueNoQueueAbortRecover)
-{
- cout << test_filename << ".MultiQueueNoQueueAbortRecover: " << flush;
- tpct.testMultiQueueNoQueueAbortRecover();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueSomeQueueCommitRecover)
-{
- cout << test_filename << ".MultiQueueSomeQueueCommitRecover: " << flush;
- tpct.testMultiQueueSomeQueueCommitRecover();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueSomeQueueAbortRecover)
-{
- cout << test_filename << ".MultiQueueSomeQueueAbortRecover: " << flush;
- tpct.testMultiQueueSomeQueueAbortRecover();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueAllQueueCommitRecover)
-{
- cout << test_filename << ".MultiQueueAllQueueCommitRecover: " << flush;
- tpct.testMultiQueueAllQueueCommitRecover();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(MultiQueueAllQueueAbortRecover)
-{
- cout << test_filename << ".MultiQueueAllQueueAbortRecover: " << flush;
- tpct.testMultiQueueAllQueueAbortRecover();
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_SUITE_END()
diff --git a/qpid/cpp/src/tests/linearstore/linearstoredirsetup.sh b/qpid/cpp/src/tests/linearstore/linearstoredirsetup.sh
index 3cad50b1c5..ef39767e9b 100755
--- a/qpid/cpp/src/tests/linearstore/linearstoredirsetup.sh
+++ b/qpid/cpp/src/tests/linearstore/linearstoredirsetup.sh
@@ -19,26 +19,37 @@
# under the License.
#
-
-STORE_DIR=/tmp
-LINEARSTOREDIR=~/RedHat/linearstore
-
-rm -rf $STORE_DIR/qls
-rm -rf $STORE_DIR/p002
-rm $STORE_DIR/p004
-
-mkdir $STORE_DIR/qls
-mkdir $STORE_DIR/p002
-touch $STORE_DIR/p004
-mkdir $STORE_DIR/qls/p001
-touch $STORE_DIR/qls/p003
-ln -s $STORE_DIR/p002 $STORE_DIR/qls/p002
-ln -s $STORE_DIR/p004 $STORE_DIR/qls/p004
-
-${LINEARSTOREDIR}/tools/src/py/linearstore/efptool.py $STORE_DIR/qls/ -a -p 1 -s 2048 -n 25
-${LINEARSTOREDIR}/tools/src/py/linearstore/efptool.py $STORE_DIR/qls/ -a -p 1 -s 512 -n 25
-${LINEARSTOREDIR}/tools/src/py/linearstore/efptool.py $STORE_DIR/qls/ -a -p 2 -s 2048 -n 25
-
+# This script sets up a test directory which contains both
+# recoverable and non-recoverable files and directories for
+# the empty file pool (EFP).
+
+# NOTE: The following is based on typical development tree paths, not installed paths
+
+BASE_DIR=${HOME}/RedHat
+STORE_DIR=${BASE_DIR}
+PYTHON_TOOLS_DIR=${BASE_DIR}/qpid/tools/src/linearstore
+export PYTHONPATH=${BASE_DIR}/qpid/python:${BASE_DIR}/qpid/extras/qmf/src/py:${BASE_DIR}/qpid/tools/src/py
+
+# Remove old dirs (if present)
+rm -rf ${STORE_DIR}/qls
+rm -rf ${STORE_DIR}/p002
+rm ${STORE_DIR}/p004
+
+# Create new dir tree and links
+mkdir ${STORE_DIR}/p002_ext
+touch ${STORE_DIR}/p004_ext
+mkdir ${STORE_DIR}/qls
+mkdir ${STORE_DIR}/qls/p001
+touch ${STORE_DIR}/qls/p003
+ln -s ${STORE_DIR}/p002_ext ${STORE_DIR}/qls/p002
+ln -s ${STORE_DIR}/p004_ext ${STORE_DIR}/qls/p004
+
+# Populate efp dirs with empty files
+${PYTHON_TOOLS_DIR}/efptool.py $STORE_DIR/qls/ -a -p 1 -s 2048 -n 25
+${PYTHON_TOOLS_DIR}/efptool.py $STORE_DIR/qls/ -a -p 1 -s 512 -n 25
+${PYTHON_TOOLS_DIR}/efptool.py $STORE_DIR/qls/ -a -p 2 -s 2048 -n 25
+
+# Show the result for information
${LINEARSTOREDIR}/tools/src/py/linearstore/efptool.py $STORE_DIR/qls/ -l
tree -la $STORE_DIR/qls
diff --git a/qpid/cpp/src/tests/linearstore/tx-test-soak.sh b/qpid/cpp/src/tests/linearstore/tx-test-soak.sh
index fa05e0a4a8..7d5581961f 100755
--- a/qpid/cpp/src/tests/linearstore/tx-test-soak.sh
+++ b/qpid/cpp/src/tests/linearstore/tx-test-soak.sh
@@ -19,7 +19,6 @@
# under the License.
#
-
# tx-test-soak
#
# Basic test methodology:
@@ -30,6 +29,8 @@
# 5. Run qpid-txtest against broker in check mode, which checks that all expected messages are present.
# 6. Wash, rinse, repeat... The number of runs is determined by ${NUM_RUNS}
+# NOTE: The following is based on typical development tree paths, not installed paths
+
NUM_RUNS=1000
BASE_DIR=${HOME}/RedHat
CMAKE_BUILD_DIR=${BASE_DIR}/q.cm
@@ -43,13 +44,18 @@ BROKER_MANAGEMENT="no" # "no" or "yes"
TRUNCATE_INTERVAL=10
MAX_DISK_PERC_USED=90
-# Consts (don't adjust these...)
+# Constants (don't adjust these)
export BASE_DIR
RELATIVE_BASE_DIR=`python -c "import os,os.path; print os.path.relpath(os.environ['BASE_DIR'], os.environ['PWD'])"`
+export PYTHONPATH=${BASE_DIR}/qpid/python:${BASE_DIR}/qpid/extras/qmf/src/py:${BASE_DIR}/qpid/tools/src/py
LOG_FILE_NAME=log.txt
QPIDD_FN=qpidd
QPIDD=${CMAKE_BUILD_DIR}/src/${QPIDD_FN}
-TXTEST=${CMAKE_BUILD_DIR}/src/tests/qpid-txtest
+TXTEST_FN=qpid-txtest
+TXTEST=${CMAKE_BUILD_DIR}/src/tests/${TXTEST_FN}
+ANALYZE_FN=qpid_qls_analyze.py
+ANALYZE=${BASE_DIR}/qpid/tools/src/py/${ANALYZE_FN}
+ANALYZE_ARGS="--efp --show-recs --stats"
QPIDD_BASE_ARGS="--load-module ${STORE_MODULE} -m ${BROKER_MANAGEMENT} --auth no --default-flow-stop-threshold 0 --default-flow-resume-threshold 0 --default-queue-limit 0 --store-dir ${BASE_DIR} --log-enable ${BROKER_LOG_LEVEL} --log-to-stderr no --log-to-stdout no"
TXTEST_INIT_STR="--init yes --transfer no --check no"
TXTEST_RUN_STR="--init no --transfer yes --check no"
@@ -181,6 +187,17 @@ check_ready_to_run() {
fi
}
+# Analyze store files
+# $1: Log suffix flag: either "A" or "B". If "A", client is started in test mode, otherwise client evaluates recovery.
+analyze_store() {
+ ${ANALYZE} ${ANALYZE_ARGS} ${BASE_DIR}/qls &> ${RESULT_DIR}/qls_analysis.$1.log
+ echo >> ${RESULT_DIR}/qls_analysis.$1.log
+ echo "----------------------------------------------------------" >> ${RESULT_DIR}/qls_analysis.$1.log
+ echo "With transactional reconsiliation:" >> ${RESULT_DIR}/qls_analysis.$1.log
+ echo >> ${RESULT_DIR}/qls_analysis.$1.log
+ ${ANALYZE} ${ANALYZE_ARGS} --txn ${BASE_DIR}/qls &>> ${RESULT_DIR}/qls_analysis.$1.log
+}
+
ulimit -c unlimited # Allow core files to be created
RESULT_BASE_DIR_SUFFIX=`date "${TIMESTAMP_FORMAT}"`
@@ -219,7 +236,8 @@ for rn in `seq ${NUM_RUNS}`; do
sleep ${RUN_TIME}
kill_process ${SIG_KILL} ${QPIDD_PID}
sleep 2
- tar -czf ${RESULT_DIR}/qls_B.tar.gz ${RELATIVE_BASE_DIR}/qls
+ analyze_store "A"
+ tar -czf ${RESULT_DIR}/qls_A.tar.gz ${RELATIVE_BASE_DIR}/qls
# === PART B: Recovery and check ===
start_broker "B"
@@ -234,11 +252,14 @@ for rn in `seq ${NUM_RUNS}`; do
kill_process ${SIG_KILL} ${PID}
sleep 2
fi
- tar -czf ${RESULT_DIR}/qls_C.tar.gz ${RELATIVE_BASE_DIR}/qls
+ analyze_store "B"
+ tar -czf ${RESULT_DIR}/qls_B.tar.gz ${RELATIVE_BASE_DIR}/qls
# === Check for errors, cores and exceptions in logs ===
grep -Hn "jexception" ${RESULT_DIR}/qpidd.A.log | tee -a ${LOG_FILE}
grep -Hn "jexception" ${RESULT_DIR}/qpidd.B.log | tee -a ${LOG_FILE}
+ grep -Hn "Traceback (most recent call last):" ${RESULT_DIR}/qls_analysis.A.log | tee -a ${LOG_FILE}
+ grep -Hn "Traceback (most recent call last):" ${RESULT_DIR}/qls_analysis.B.log | tee -a ${LOG_FILE}
grep "${SUCCESS_MSG}" ${RESULT_DIR}/txtest.B.log &> /dev/null
if [[ "$?" != "0" ]]; then
echo "ERROR in run ${rn}" >> ${LOG_FILE}
View Lorry Controller Status