[rhmessaging-commits] rhmessaging commits: r3913 - store/trunk/cpp/tests.
rhmessaging-commits at lists.jboss.org
rhmessaging-commits at lists.jboss.org
Thu Apr 15 11:58:33 EDT 2010
Author: kpvdr
Date: 2010-04-15 11:58:32 -0400 (Thu, 15 Apr 2010)
New Revision: 3913
Modified:
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:
Implementation of QPID-2509 (Remove message staging from C++ broker) - syncs with qpid r.934463
Modified: store/trunk/cpp/tests/OrderingTest.cpp
===================================================================
--- store/trunk/cpp/tests/OrderingTest.cpp 2010-04-14 17:34:58 UTC (rev 3912)
+++ store/trunk/cpp/tests/OrderingTest.cpp 2010-04-15 15:58:32 UTC (rev 3913)
@@ -105,7 +105,7 @@
sys::Timer t;
DtxManager mgr(t);
mgr.setStore (store.get());
- RecoveryManagerImpl recoveryMgr(queues, exchanges, links, mgr, 0);
+ RecoveryManagerImpl recoveryMgr(queues, exchanges, links, mgr);
store->recover(recoveryMgr);
queue = queues.find(name);
Modified: store/trunk/cpp/tests/SimpleTest.cpp
===================================================================
--- store/trunk/cpp/tests/SimpleTest.cpp 2010-04-14 17:34:58 UTC (rev 3912)
+++ store/trunk/cpp/tests/SimpleTest.cpp 2010-04-15 15:58:32 UTC (rev 3913)
@@ -70,7 +70,7 @@
sys::Timer t;
DtxManager mgr(t);
mgr.setStore (&store);
- RecoveryManagerImpl recovery(queues, exchanges, links, mgr, 0);
+ RecoveryManagerImpl recovery(queues, exchanges, links, mgr);
store.recover(recovery);
}
@@ -360,136 +360,6 @@
cout << "ok" << endl;
}
-QPID_AUTO_TEST_CASE(Staging)
-{
- cout << test_filename << ".Staging: " << flush;
-
- const string name("MyDurableQueue");
- const string exchange("MyExchange");
- const string routingKey("MyRoutingKey");
- const Uuid messageId(true);
- const string data1("abcdefghijklmnopqrstuvwxyz");
- const string data2("ABCDEFGHIJKLMNOPQRSTUVWXYZ");
- {
- MessageStoreImpl store;
- store.init(test_dir, 4, 1, true); // truncate store
-
- //create & stage a message
- boost::intrusive_ptr<Message> msg = MessageUtils::createMessage(exchange, routingKey, messageId, (data1.size() + data2.size()));
- intrusive_ptr<PersistableMessage> pmsg = static_pointer_cast<PersistableMessage>(msg);
- intrusive_ptr<const PersistableMessage> cpmsg = static_pointer_cast<const PersistableMessage>(msg);
- msg->getProperties<DeliveryProperties>()->setDeliveryMode(PERSISTENT);
- FieldTable table;
- table.setString("abc", "xyz");
- msg->getProperties<MessageProperties>()->setApplicationHeaders(table);
- store.stage(pmsg);
-
- //append to it
- msg->setStore(&store);
- msg->releaseContent();//ensure that data is not held in memory but is appended to disk when added
- store.appendContent(cpmsg, data1);
- store.appendContent(cpmsg, data2);
-
- //enqueue it
- Queue::shared_ptr queue(new Queue(name, 0, &store, 0));
- FieldTable settings;
- queue->create(settings);
- queue->enqueue(0, msg);
-
- //load it (without recovery)
- DummyHandler handler;
- QueuedMessage qm(queue.get(), msg, 0);
- //52 chars of data, i.e. 2 chunks of 20 and one of 12
- MessageUtils::deliver(qm, handler, 20);
- BOOST_REQUIRE(handler.frames.size() > 1);
- string loaded;
- for (uint i = 1; i < handler.frames.size(); i++) {
- AMQContentBody* contentBody(dynamic_cast<AMQContentBody*>(handler.frames[i].getBody()));
- BOOST_REQUIRE(contentBody);
- loaded += contentBody->getData();
- }
- BOOST_CHECK_EQUAL(data1 + data2, loaded);
-
- }//db will be closed
- {
- //recover
- MessageStoreImpl store;
- store.init(test_dir, 4, 1);
- QueueRegistry registry;
- registry.setStore (&store);
- ExchangeRegistry exchanges;
- LinkRegistry links;
- sys::Timer t;
- DtxManager dtx(t);
- dtx.setStore (&store);
- RecoveryManagerImpl recovery(registry, exchanges, links, dtx, 10);
- store.recover(recovery);
-
- //get message instance from queue
- Queue::shared_ptr queue = registry.find(name);
- BOOST_REQUIRE(queue);
- BOOST_CHECK_EQUAL((u_int32_t) 1, queue->getMessageCount());
- boost::intrusive_ptr<Message> msg = queue->get().payload;
-
- //check headers
- BOOST_CHECK_EQUAL(exchange, msg->getExchangeName());
- BOOST_CHECK_EQUAL(routingKey, msg->getRoutingKey());
- BOOST_CHECK_EQUAL(messageId, msg->getProperties<MessageProperties>()->getMessageId());
- BOOST_CHECK_EQUAL((uint8_t) PERSISTENT, msg->getProperties<DeliveryProperties>()->getDeliveryMode());
- BOOST_REQUIRE(Str16Value("xyz") == *msg->getProperties<MessageProperties>()->getApplicationHeaders().get("abc"));
- BOOST_CHECK_EQUAL((u_int64_t) (data1.size() + data2.size()), msg->getFrames().getHeaders()->getContentLength());
-
- BOOST_CHECK_EQUAL((u_int64_t) 0, msg->contentSize());//ensure it is being lazily loaded
-
- //load lazily
- DummyHandler handler;
- QueuedMessage qm(queue.get(),msg,0);
- //52 chars of data, i.e. 2 chunks of 20 and one of 12
- MessageUtils::deliver(qm, handler, 20);
-
- BOOST_REQUIRE(handler.frames.size() > 1);
- string loaded;
- for (uint i = 1; i < handler.frames.size(); i++) {
- AMQContentBody* contentBody(dynamic_cast<AMQContentBody*>(handler.frames[i].getBody()));
- BOOST_REQUIRE(contentBody);
- loaded += contentBody->getData();
- }
- BOOST_CHECK_EQUAL(data1 + data2, loaded);
-
- //dequeue
- queue->dequeue(0, qm);
- }
-
- cout << "ok" << endl;
-}
-
-QPID_AUTO_TEST_CASE(DestroyStagedMessage)
-{
- cout << test_filename << ".DestroyStagedMessage: " << flush;
-
- MessageStoreImpl store;
- store.init(test_dir, 4, 1, true); // truncate store
-
- const string data("abcdefg");
- boost::intrusive_ptr<Message> msg(MessageUtils::createMessage("my_exchange", "my_routing_key", "my_message", data.length()));
- intrusive_ptr<PersistableMessage> pmsg = static_pointer_cast<PersistableMessage>(msg);
- intrusive_ptr<const PersistableMessage> cpmsg = static_pointer_cast<const PersistableMessage>(msg);
- MessageUtils::addContent(msg, data);
-
- store.stage(pmsg);
- store.destroy(*pmsg);
-
- try {
- string loaded;
- Queue queue("dummy", 0, &store, 0);
- store.loadContent(queue, cpmsg, loaded, 0, data.length());
- BOOST_FAIL("store.loadContent() did not throw StoreException as expected.");
- } catch (StoreException& e) {
- }
-
- cout << "ok" << endl;
-}
-
QPID_AUTO_TEST_CASE(DestroyEnqueuedMessage)
{
cout << test_filename << ".DestroyEnqueuedMessage: " << flush;
Modified: store/trunk/cpp/tests/TransactionalTest.cpp
===================================================================
--- store/trunk/cpp/tests/TransactionalTest.cpp 2010-04-14 17:34:58 UTC (rev 3912)
+++ store/trunk/cpp/tests/TransactionalTest.cpp 2010-04-15 15:58:32 UTC (rev 3913)
@@ -136,7 +136,7 @@
sys::Timer t;
DtxManager mgr(t);
mgr.setStore (store.get());
- RecoveryManagerImpl recovery(*queues, exchanges, links, mgr, 0);
+ RecoveryManagerImpl recovery(*queues, exchanges, links, mgr);
store->recover(recovery);
queueA = queues->find(nameA);
Modified: store/trunk/cpp/tests/TwoPhaseCommitTest.cpp
===================================================================
--- store/trunk/cpp/tests/TwoPhaseCommitTest.cpp 2010-04-14 17:34:58 UTC (rev 3912)
+++ store/trunk/cpp/tests/TwoPhaseCommitTest.cpp 2010-04-15 15:58:32 UTC (rev 3913)
@@ -361,7 +361,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, 0);
+ RecoveryManagerImpl recovery(*queues, exchanges, *links, *dtxmgr);
store->recover(recovery);
queueA = queues->find(nameA);
More information about the rhmessaging-commits
mailing list