rhmessaging commits: r4522 - in store/trunk/cpp: tests and 1 other directory.
by rhmessaging-commits＠lists.jboss.org
Author: kpvdr
Date: 2012-12-05 11:32:41 -0500 (Wed, 05 Dec 2012)
New Revision: 4522
Modified:
store/trunk/cpp/lib/MessageStoreImpl.cpp
store/trunk/cpp/lib/MessageStoreImpl.h
store/trunk/cpp/lib/StorePlugin.cpp
store/trunk/cpp/tests/OrderingTest.cpp
store/trunk/cpp/tests/SimpleTest.cpp
store/trunk/cpp/tests/TransactionalTest.cpp
store/trunk/cpp/tests/TwoPhaseCommitTest.cpp
Log:
QPID-4490 / BZ873674 - Durable mesages with TTL do not expire after broker recovery (c++ store)
Modified: store/trunk/cpp/lib/MessageStoreImpl.cpp
===================================================================
--- store/trunk/cpp/lib/MessageStoreImpl.cpp 2012-11-30 19:18:38 UTC (rev 4521)
+++ store/trunk/cpp/lib/MessageStoreImpl.cpp 2012-12-05 16:32:41 UTC (rev 4522)
@@ -59,7 +59,7 @@
tpc_flag(_tpc_flag)
{}
-MessageStoreImpl::MessageStoreImpl(qpid::sys::Timer& timer_, const char* envpath) :
+MessageStoreImpl::MessageStoreImpl(qpid::broker::Broker* broker_, const char* envpath) :
numJrnlFiles(0),
autoJrnlExpand(false),
autoJrnlExpandMaxFiles(0),
@@ -74,7 +74,7 @@
highestRid(0),
isInit(false),
envPath(envpath),
- timer(timer_),
+ broker(broker_),
agent(0)
{}
@@ -219,7 +219,7 @@
autoJrnlExpandMaxFiles = p;
}
-void MessageStoreImpl::initManagement (qpid::broker::Broker* broker)
+void MessageStoreImpl::initManagement ()
{
if (broker != 0) {
agent = broker->getManagementAgent();
@@ -364,7 +364,7 @@
// NOTE: during normal initialization, agent == 0 because the store is initialized before the management infrastructure.
// However during a truncated initialization in a cluster, agent != 0. We always pass 0 as the agent for the
// TplStore to keep things consistent in a cluster. See https://bugzilla.redhat.com/show_bug.cgi?id=681026
- tplStorePtr.reset(new TplJournalImpl(timer, "TplStore", getTplBaseDir(), "tpl", defJournalGetEventsTimeout, defJournalFlushTimeout, 0));
+ tplStorePtr.reset(new TplJournalImpl(broker->getTimer(), "TplStore", getTplBaseDir(), "tpl", defJournalGetEventsTimeout, defJournalFlushTimeout, 0));
isInit = true;
} catch (const DbException& e) {
if (e.get_errno() == DB_VERSION_MISMATCH)
@@ -515,7 +515,7 @@
return;
}
- jQueue = new JournalImpl(timer, queue.getName(), getJrnlDir(queue), std::string("JournalData"),
+ jQueue = new JournalImpl(broker->getTimer(), queue.getName(), getJrnlDir(queue), std::string("JournalData"),
defJournalGetEventsTimeout, defJournalFlushTimeout, agent,
boost::bind(&MessageStoreImpl::journalDeleted, this, _1));
{
@@ -801,7 +801,7 @@
QPID_LOG(error, "Cannot recover empty (null) queue name - ignoring and attempting to continue.");
break;
}
- jQueue = new JournalImpl(timer, queueName, getJrnlHashDir(queueName), std::string("JournalData"),
+ jQueue = new JournalImpl(broker->getTimer(), queueName, getJrnlHashDir(queueName), std::string("JournalData"),
defJournalGetEventsTimeout, defJournalFlushTimeout, agent,
boost::bind(&MessageStoreImpl::journalDeleted, this, _1));
{
@@ -981,6 +981,8 @@
// At some future point if delivery attempts are stored, then this call would
// become optional depending on that information.
msg->setRedelivered();
+ // Reset the TTL for the recovered message
+ msg->computeExpiration(broker->getExpiryPolicy());
u_int32_t contentOffset = headerSize + preambleLength;
u_int64_t contentSize = readSize - contentOffset;
Modified: store/trunk/cpp/lib/MessageStoreImpl.h
===================================================================
--- store/trunk/cpp/lib/MessageStoreImpl.h 2012-11-30 19:18:38 UTC (rev 4521)
+++ store/trunk/cpp/lib/MessageStoreImpl.h 2012-12-05 16:32:41 UTC (rev 4522)
@@ -152,7 +152,7 @@
u_int64_t highestRid;
bool isInit;
const char* envPath;
- qpid::sys::Timer& timer;
+ qpid::broker::Broker* broker;
qmf::com::redhat::rhm::store::Store::shared_ptr mgmtObject;
qpid::management::ManagementAgent* agent;
@@ -275,7 +275,7 @@
public:
typedef boost::shared_ptr<MessageStoreImpl> shared_ptr;
- MessageStoreImpl(qpid::sys::Timer& timer, const char* envpath = 0);
+ MessageStoreImpl(qpid::broker::Broker* broker, const char* envpath = 0);
virtual ~MessageStoreImpl();
@@ -294,7 +294,7 @@
void truncateInit(const bool saveStoreContent = false);
- void initManagement (qpid::broker::Broker* broker);
+ void initManagement ();
void finalize();
Modified: store/trunk/cpp/lib/StorePlugin.cpp
===================================================================
--- store/trunk/cpp/lib/StorePlugin.cpp 2012-11-30 19:18:38 UTC (rev 4521)
+++ store/trunk/cpp/lib/StorePlugin.cpp 2012-12-05 16:32:41 UTC (rev 4522)
@@ -46,7 +46,7 @@
{
Broker* broker = dynamic_cast<Broker*>(&target);
if (!broker) return;
- store.reset(new MessageStoreImpl(broker->getTimer()));
+ store.reset(new MessageStoreImpl(broker));
DataDir& dataDir = broker->getDataDir ();
if (options.storeDir.empty ())
{
@@ -71,7 +71,7 @@
QPID_LOG(info, "Disabling management instrumentation for the store in a cluster.");
} else {
QPID_LOG(info, "Enabling management instrumentation for the store.");
- store->initManagement(broker);
+ store->initManagement();
}
}
Modified: store/trunk/cpp/tests/OrderingTest.cpp
===================================================================
--- store/trunk/cpp/tests/OrderingTest.cpp 2012-11-30 19:18:38 UTC (rev 4521)
+++ store/trunk/cpp/tests/OrderingTest.cpp 2012-12-05 16:32:41 UTC (rev 4522)
@@ -32,8 +32,8 @@
#include "qpid/log/Logger.h"
#include "qpid/sys/Timer.h"
-qpid::sys::Timer timer;
-qpid::broker::ProtocolRegistry pr;
+qpid::broker::Broker::Options opts;
+qpid::broker::Broker br(opts);
#define SET_LOG_LEVEL(level) \
qpid::log::Options opts(""); \
@@ -84,7 +84,7 @@
void setup()
{
- store = std::auto_ptr<MessageStoreImpl>(new MessageStoreImpl(timer));
+ 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));
@@ -112,14 +112,14 @@
queue.reset();
store.reset();
- store = std::auto_ptr<MessageStoreImpl>(new MessageStoreImpl(timer));
+ 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());
- RecoveryManagerImpl recoveryMgr(queues, exchanges, links, mgr, pr);
+ RecoveryManagerImpl recoveryMgr(queues, exchanges, links, mgr, br.getProtocolRegistry());
store->recover(recoveryMgr);
queue = queues.find(name);
Modified: store/trunk/cpp/tests/SimpleTest.cpp
===================================================================
--- store/trunk/cpp/tests/SimpleTest.cpp 2012-11-30 19:18:38 UTC (rev 4521)
+++ store/trunk/cpp/tests/SimpleTest.cpp 2012-12-05 16:32:41 UTC (rev 4522)
@@ -36,8 +36,8 @@
#include "qpid/log/Logger.h"
#include "qpid/sys/Timer.h"
-qpid::sys::Timer timer;
-qpid::broker::ProtocolRegistry pr;
+qpid::broker::Broker::Options opts;
+qpid::broker::Broker br(opts);
#define SET_LOG_LEVEL(level) \
qpid::log::Options opts(""); \
@@ -76,7 +76,7 @@
sys::Timer t;
DtxManager mgr(t);
mgr.setStore (&store);
- RecoveryManagerImpl recovery(queues, exchanges, links, mgr, pr);
+ RecoveryManagerImpl recovery(queues, exchanges, links, mgr, br.getProtocolRegistry());
store.recover(recovery);
}
@@ -98,7 +98,7 @@
const string& key, const FieldTable& args)
{
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1, true); // truncate store
Exchange::shared_ptr exchange(new DirectExchange(exchangeName, true, args));
Queue::shared_ptr queue(new Queue(queueName, 0, &store, 0));
@@ -108,7 +108,7 @@
store.bind(*exchange, *queue, key, args);
}//db will be closed
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
ExchangeRegistry exchanges;
QueueRegistry queues;
@@ -127,7 +127,7 @@
store.unbind(*exchange, *queue, key, args);
}
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
ExchangeRegistry exchanges;
QueueRegistry queues;
@@ -154,7 +154,7 @@
SET_LOG_LEVEL("error+"); // This only needs to be set once.
cout << test_filename << ".CreateDelete: " << flush;
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1, true); // truncate store
string name("CreateDeleteQueue");
Queue queue(name, 0, &store, 0);
@@ -170,7 +170,7 @@
QPID_AUTO_TEST_CASE(EmptyRecover)
{
cout << test_filename << ".EmptyRecover: " << flush;
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1, true); // truncate store
QueueRegistry registry;
registry.setStore (&store);
@@ -187,7 +187,7 @@
uint64_t id(0);
string name("MyDurableQueue");
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1, true); // truncate store
Queue queue(name, 0, &store, 0);
store.create(queue, qpid::framing::FieldTable());
@@ -195,7 +195,7 @@
id = queue.getPersistenceId();
}//db will be closed
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
QueueRegistry registry;
registry.setStore (&store);
@@ -219,14 +219,14 @@
settings.populate(arguments, settings.storeSettings);
string name("MyDurableQueue");
{
- MessageStoreImpl store(timer);
+ 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(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
QueueRegistry registry;
registry.setStore (&store);
@@ -248,14 +248,14 @@
string name("MyDurableQueue");
{
- MessageStoreImpl store(timer);
+ 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(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
QueueRegistry registry;
registry.setStore (&store);
@@ -281,7 +281,7 @@
string data1("abcdefg");
string data2("hijklmn");
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1, true); // truncate store
Queue::shared_ptr queue(new Queue(name, 0, &store, 0));
queue->create();
@@ -295,7 +295,7 @@
queue->deliver(msg);
}//db will be closed
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
QueueRegistry registry;
registry.setStore (&store);
@@ -333,7 +333,7 @@
string routingKey("MyRoutingKey");
Uuid messageId(true);
string data("abcdefg");
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1, true); // truncate store
Queue::shared_ptr queue(new Queue(name, 0, &store, 0));
queue->create();
@@ -348,7 +348,7 @@
queue->dequeue(0, cursor);
}//db will be closed
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
QueueRegistry registry;
registry.setStore (&store);
@@ -371,7 +371,7 @@
FieldTable args;
args.setString("a", "A");
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1, true); // truncate store
ExchangeRegistry registry;
Exchange::shared_ptr exchange = registry.declare(name, type, true, args).first;
@@ -380,7 +380,7 @@
BOOST_REQUIRE(id);
}//db will be closed
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
ExchangeRegistry registry;
@@ -394,7 +394,7 @@
store.destroy(*exchange);
}
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
ExchangeRegistry registry;
@@ -442,7 +442,7 @@
string key("my-routing-key");
FieldTable args;
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1, true); // truncate store
Exchange::shared_ptr exchange(new DirectExchange(exchangeName, true, args));
Queue::shared_ptr queue1(new Queue(queueName1, 0, &store, 0));
@@ -456,7 +456,7 @@
store.destroy(*queue1);
}//db will be closed
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
ExchangeRegistry exchanges;
QueueRegistry queues;
@@ -473,7 +473,7 @@
store.destroy(*exchange);
}
{
- MessageStoreImpl store(timer);
+ MessageStoreImpl store(&br);
store.init(test_dir, 4, 1);
ExchangeRegistry exchanges;
QueueRegistry queues;
Modified: store/trunk/cpp/tests/TransactionalTest.cpp
===================================================================
--- store/trunk/cpp/tests/TransactionalTest.cpp 2012-11-30 19:18:38 UTC (rev 4521)
+++ store/trunk/cpp/tests/TransactionalTest.cpp 2012-12-05 16:32:41 UTC (rev 4522)
@@ -34,8 +34,8 @@
#include "qpid/log/Logger.h"
#include "qpid/sys/Timer.h"
-qpid::sys::Timer timer;
-qpid::broker::ProtocolRegistry pr;
+qpid::broker::Broker::Options opts;
+qpid::broker::Broker br(opts);
#define SET_LOG_LEVEL(level) \
qpid::log::Options opts(""); \
@@ -73,7 +73,7 @@
class TestMessageStore: public MessageStoreImpl
{
public:
- TestMessageStore(qpid::sys::Timer& timer, const char* envpath = 0) : MessageStoreImpl(timer, envpath) {}
+ TestMessageStore(qpid::broker::Broker* br, const char* envpath = 0) : MessageStoreImpl(br, envpath) {}
std::auto_ptr<qpid::broker::TransactionContext> begin() {
checkInit();
// pass sequence number for c/a
@@ -113,7 +113,7 @@
template <class T>
void setup()
{
- store = std::auto_ptr<T>(new T(timer));
+ store = std::auto_ptr<T>(new T(&br));
store->init(test_dir, 4, 1, true); // truncate store
//create two queues:
@@ -131,7 +131,7 @@
queues.reset();
store.reset();
- store = std::auto_ptr<T>(new T(timer));
+ store = std::auto_ptr<T>(new T(&br));
store->init(test_dir, 4, 1);
queues = std::auto_ptr<QueueRegistry>(new QueueRegistry);
ExchangeRegistry exchanges;
@@ -139,7 +139,7 @@
sys::Timer t;
DtxManager mgr(t);
mgr.setStore (store.get());
- RecoveryManagerImpl recovery(*queues, exchanges, links, mgr, pr);
+ RecoveryManagerImpl recovery(*queues, exchanges, links, mgr, br.getProtocolRegistry());
store->recover(recovery);
queueA = queues->find(nameA);
Modified: store/trunk/cpp/tests/TwoPhaseCommitTest.cpp
===================================================================
--- store/trunk/cpp/tests/TwoPhaseCommitTest.cpp 2012-11-30 19:18:38 UTC (rev 4521)
+++ store/trunk/cpp/tests/TwoPhaseCommitTest.cpp 2012-12-05 16:32:41 UTC (rev 4522)
@@ -34,8 +34,8 @@
#include "qpid/log/Logger.h"
#include "qpid/sys/Timer.h"
-qpid::sys::Timer timer;
-qpid::broker::ProtocolRegistry pr;
+qpid::broker::Broker::Options opts;
+qpid::broker::Broker br(opts);
#define SET_LOG_LEVEL(level) \
qpid::log::Options opts(""); \
@@ -186,7 +186,7 @@
class TestMessageStore: public MessageStoreImpl
{
public:
- TestMessageStore(qpid::sys::Timer& timer, const char* envpath = 0) : MessageStoreImpl(timer, envpath) {}
+ TestMessageStore(qpid::broker::Broker* br, const char* envpath = 0) : MessageStoreImpl(br, envpath) {}
std::auto_ptr<qpid::broker::TPCTransactionContext> begin(const std::string& xid) {
checkInit();
IdSequence* jtx = &messageIdSequence;
@@ -352,7 +352,7 @@
template <class T>
void setup()
{
- store = std::auto_ptr<T>(new T(timer));
+ store = std::auto_ptr<T>(new T(&br));
store->init(test_dir, 4, 1, true); // truncate store
//create two queues:
@@ -377,7 +377,7 @@
queues.reset();
links.reset();
- store = std::auto_ptr<T>(new T(timer));
+ store = std::auto_ptr<T>(new T(&br));
store->init(test_dir, 4, 1);
sys::Timer t;
ExchangeRegistry exchanges;
@@ -385,7 +385,7 @@
links = std::auto_ptr<LinkRegistry>(new LinkRegistry);
dtxmgr = std::auto_ptr<DtxManager>(new DtxManager(t));
dtxmgr->setStore (store.get());
- RecoveryManagerImpl recovery(*queues, exchanges, *links, *dtxmgr, pr);
+ RecoveryManagerImpl recovery(*queues, exchanges, *links, *dtxmgr, br.getProtocolRegistry());
store->recover(recovery);
queueA = queues->find(nameA);
12 years
- 1
- 0
- ← Newer
- 1
- Older →