You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2007/11/14 22:02:23 UTC
svn commit: r595056 - in /incubator/qpid/trunk/qpid/cpp/src: qpid/broker/
qpid/cluster/ qpid/management/ tests/
Author: aconway
Date: Wed Nov 14 13:02:14 2007
New Revision: 595056
URL: http://svn.apache.org/viewvc?rev=595056&view=rev
Log:
Replaced shared_ptr with intrusive_ptr for qpid::Broker::Message.
Gives 9% reduction in broker heap use (perftest --count 100000.)
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageDelivery.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Persistable.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.h
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/MessageUtils.h
incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/TxAckTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/TxPublishTest.cpp
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Consumer.h Wed Nov 14 13:02:14 2007
@@ -33,11 +33,11 @@
struct QueuedMessage
{
- Message::shared_ptr payload;
+ intrusive_ptr<Message> payload;
framing::SequenceNumber position;
Queue* queue;
- QueuedMessage(Queue* q, Message::shared_ptr msg, framing::SequenceNumber sn) :
+ QueuedMessage(Queue* q, intrusive_ptr<Message> msg, framing::SequenceNumber sn) :
payload(msg), position(sn), queue(q) {}
QueuedMessage(Queue* q) : queue(q) {}
};
@@ -53,7 +53,7 @@
Consumer(bool preAcquires = true) : acquires(preAcquires) {}
bool preAcquires() const { return acquires; }
virtual bool deliver(QueuedMessage& msg) = 0;
- virtual bool filter(Message::shared_ptr) { return true; }
+ virtual bool filter(intrusive_ptr<Message>) { return true; }
virtual ~Consumer(){}
};
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.cpp Wed Nov 14 13:02:14 2007
@@ -22,7 +22,7 @@
using namespace qpid::broker;
-DeliverableMessage::DeliverableMessage(Message::shared_ptr& _msg) : msg(_msg)
+DeliverableMessage::DeliverableMessage(intrusive_ptr<Message>& _msg) : msg(_msg)
{
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DeliverableMessage.h Wed Nov 14 13:02:14 2007
@@ -28,9 +28,9 @@
namespace qpid {
namespace broker {
class DeliverableMessage : public Deliverable{
- Message::shared_ptr msg;
+ intrusive_ptr<Message> msg;
public:
- DeliverableMessage(Message::shared_ptr& msg);
+ DeliverableMessage(intrusive_ptr<Message>& msg);
virtual void deliverTo(Queue::shared_ptr& queue);
Message& getMessage();
virtual ~DeliverableMessage(){}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.cpp Wed Nov 14 13:02:14 2007
@@ -77,7 +77,7 @@
completed.update(command, command);
}
-void IncomingExecutionContext::track(Message::shared_ptr msg)
+void IncomingExecutionContext::track(intrusive_ptr<Message> msg)
{
if (msg->isEnqueueComplete()) {
complete(msg->getCommandId());
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncomingExecutionContext.h Wed Nov 14 13:02:14 2007
@@ -30,7 +30,7 @@
class IncomingExecutionContext
{
- typedef std::list<Message::shared_ptr> Messages;
+ typedef std::list<intrusive_ptr<Message> > Messages;
framing::Window window;
framing::AccumulatedAck completed;
Messages incomplete;
@@ -45,7 +45,7 @@
void sync(const framing::SequenceNumber& point);
framing::SequenceNumber next();
void complete(const framing::SequenceNumber& command);
- void track(Message::shared_ptr);
+ void track(intrusive_ptr<Message>);
const framing::SequenceNumber& getMark();
framing::SequenceNumberSet getRange();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h Wed Nov 14 13:02:14 2007
@@ -45,7 +45,7 @@
class Message : public PersistableMessage {
public:
- typedef boost::shared_ptr<Message> shared_ptr;
+ typedef boost::intrusive_ptr<Message> shared_ptr;
Message(const framing::SequenceNumber& id = framing::SequenceNumber());
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.cpp Wed Nov 14 13:02:14 2007
@@ -62,14 +62,14 @@
void MessageBuilder::end()
{
- message.reset();
+ message = 0;
state = DORMANT;
staging = false;
}
void MessageBuilder::start(const SequenceNumber& id)
{
- message = Message::shared_ptr(new Message(id));
+ message = intrusive_ptr<Message>(new Message(id));
state = METHOD;
staging = false;
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageBuilder.h Wed Nov 14 13:02:14 2007
@@ -21,9 +21,9 @@
#ifndef _MessageBuilder_
#define _MessageBuilder_
-#include "boost/shared_ptr.hpp"
#include "qpid/framing/FrameHandler.h"
#include "qpid/framing/SequenceNumber.h"
+#include "qpid/RefCounted.h"
namespace qpid {
namespace broker {
@@ -34,13 +34,13 @@
public:
MessageBuilder(MessageStore* const store = 0, uint64_t stagingThreshold = 0);
void handle(framing::AMQFrame& frame);
- boost::shared_ptr<Message> getMessage() { return message; }
+ intrusive_ptr<Message> getMessage() { return message; }
void start(const framing::SequenceNumber& id);
void end();
private:
enum State {DORMANT, METHOD, HEADER, CONTENT};
State state;
- boost::shared_ptr<Message> message;
+ intrusive_ptr<Message> message;
MessageStore* const store;
const uint64_t stagingThreshold;
bool staging;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageDelivery.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageDelivery.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageDelivery.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageDelivery.cpp Wed Nov 14 13:02:14 2007
@@ -39,7 +39,7 @@
struct BaseToken : DeliveryToken
{
virtual ~BaseToken() {}
- virtual AMQFrame sendMethod(Message::shared_ptr msg, DeliveryId id) = 0;
+ virtual AMQFrame sendMethod(intrusive_ptr<Message> msg, DeliveryId id) = 0;
};
struct BasicGetToken : BaseToken
@@ -50,7 +50,7 @@
BasicGetToken(Queue::shared_ptr q) : queue(q) {}
- AMQFrame sendMethod(Message::shared_ptr msg, DeliveryId id)
+ AMQFrame sendMethod(intrusive_ptr<Message> msg, DeliveryId id)
{
return AMQFrame(0, BasicGetOkBody(
ProtocolVersion(), id.getValue(), msg->getRedelivered(), msg->getExchangeName(),
@@ -66,7 +66,7 @@
BasicConsumeToken(const string c) : consumer(c) {}
- AMQFrame sendMethod(Message::shared_ptr msg, DeliveryId id)
+ AMQFrame sendMethod(intrusive_ptr<Message> msg, DeliveryId id)
{
return AMQFrame(0, BasicDeliverBody(
ProtocolVersion(), consumer, id.getValue(),
@@ -84,7 +84,7 @@
MessageDeliveryToken(const std::string& d, u_int8_t c, u_int8_t a) :
destination(d), confirmMode(c), acquireMode(a) {}
- AMQFrame sendMethod(Message::shared_ptr msg, DeliveryId /*id*/)
+ AMQFrame sendMethod(intrusive_ptr<Message> msg, DeliveryId /*id*/)
{
//may need to set the redelivered flag:
if (msg->getRedelivered()){
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Persistable.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Persistable.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Persistable.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Persistable.h Wed Nov 14 13:02:14 2007
@@ -24,6 +24,7 @@
#include "qpid/framing/amqp_types.h"
#include "qpid/framing/Buffer.h"
+#include "qpid/RefCounted.h"
namespace qpid {
namespace broker {
@@ -31,7 +32,7 @@
/**
* Base class for all persistable objects
*/
-class Persistable
+class Persistable : public RefCounted
{
public:
/**
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Wed Nov 14 13:02:14 2007
@@ -81,7 +81,7 @@
}
-void Queue::deliver(Message::shared_ptr& msg){
+void Queue::deliver(intrusive_ptr<Message>& msg){
if (msg->isImmediate() && getConsumerCount() == 0) {
if (alternateExchange) {
DeliverableMessage deliverable(msg);
@@ -108,7 +108,7 @@
}
-void Queue::recover(Message::shared_ptr& msg){
+void Queue::recover(intrusive_ptr<Message>& msg){
push(msg);
msg->enqueueComplete(); // mark the message as enqueued
if (mgmtObject != 0)
@@ -120,7 +120,7 @@
}
}
-void Queue::process(Message::shared_ptr& msg){
+void Queue::process(intrusive_ptr<Message>& msg){
uint32_t mask = management::MSG_MASK_TX;
@@ -178,7 +178,7 @@
* the message, or if the queue is exclusive to a single connection
* and has a single consumer (covers the JMS topic case).
*/
-bool Queue::exclude(Message::shared_ptr msg)
+bool Queue::exclude(intrusive_ptr<Message> msg)
{
RWlock::ScopedWlock locker(consumerLock);
if (exclusive) {
@@ -373,7 +373,7 @@
messages.pop_front();
}
-void Queue::push(Message::shared_ptr& msg){
+void Queue::push(intrusive_ptr<Message>& msg){
Mutex::ScopedLock locker(messageLock);
messages.push_back(QueuedMessage(this, msg, ++sequence));
if (policy.get()) {
@@ -412,7 +412,7 @@
}
// return true if store exists,
-bool Queue::enqueue(TransactionContext* ctxt, Message::shared_ptr msg)
+bool Queue::enqueue(TransactionContext* ctxt, intrusive_ptr<Message> msg)
{
if (msg->isPersistent() && store) {
msg->enqueueAsync(this, store); //increment to async counter -- for message sent to more than one queue
@@ -423,7 +423,7 @@
}
// return true if store exists,
-bool Queue::dequeue(TransactionContext* ctxt, Message::shared_ptr msg)
+bool Queue::dequeue(TransactionContext* ctxt, intrusive_ptr<Message> msg)
{
if (msg->isPersistent() && store) {
msg->dequeueAsync(this, store); //increment to async counter -- for message sent to more than one queue
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Wed Nov 14 13:02:14 2007
@@ -99,7 +99,7 @@
management::Queue::shared_ptr mgmtObject;
void pop();
- void push(Message::shared_ptr& msg);
+ void push(intrusive_ptr<Message>& msg);
bool dispatch(QueuedMessage& msg);
void setPolicy(std::auto_ptr<QueuePolicy> policy);
/**
@@ -113,7 +113,7 @@
bool seek(QueuedMessage& msg, const framing::SequenceNumber& position);
uint32_t getAcquirerCount() const;
bool getNextMessage(QueuedMessage& msg);
- bool exclude(Message::shared_ptr msg);
+ bool exclude(intrusive_ptr<Message> msg);
public:
@@ -140,12 +140,12 @@
* Delivers a message to the queue. Will record it as
* enqueued if persistent then process it.
*/
- void deliver(Message::shared_ptr& msg);
+ void deliver(intrusive_ptr<Message>& msg);
/**
* Dispatches the messages immediately to a consumer if
* one is available or stores it for later if not.
*/
- void process(Message::shared_ptr& msg);
+ void process(intrusive_ptr<Message>& msg);
/**
* Returns a message to the in-memory queue (due to lack
* of acknowledegement from a receiver). If a consumer is
@@ -156,7 +156,7 @@
/**
* Used during recovery to add stored messages back to the queue
*/
- void recover(Message::shared_ptr& msg);
+ void recover(intrusive_ptr<Message>& msg);
/**
* Request dispatch any queued messages providing there are
* consumers for them. Only one thread can be dispatching
@@ -181,11 +181,11 @@
inline bool isAutoDelete() const { return autodelete; }
bool canAutoDelete() const;
- bool enqueue(TransactionContext* ctxt, Message::shared_ptr msg);
+ bool enqueue(TransactionContext* ctxt, intrusive_ptr<Message> msg);
/**
* dequeue from store (only done once messages is acknowledged)
*/
- bool dequeue(TransactionContext* ctxt, Message::shared_ptr msg);
+ bool dequeue(TransactionContext* ctxt, intrusive_ptr<Message> msg);
/**
* dequeues from memory only
*/
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.cpp Wed Nov 14 13:02:14 2007
@@ -22,7 +22,7 @@
using namespace qpid::broker;
-RecoveredDequeue::RecoveredDequeue(Queue::shared_ptr _queue, Message::shared_ptr _msg) : queue(_queue), msg(_msg) {}
+RecoveredDequeue::RecoveredDequeue(Queue::shared_ptr _queue, intrusive_ptr<Message> _msg) : queue(_queue), msg(_msg) {}
bool RecoveredDequeue::prepare(TransactionContext*) throw(){
//should never be called; transaction has already prepared if an enqueue is recovered
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredDequeue.h Wed Nov 14 13:02:14 2007
@@ -34,10 +34,10 @@
namespace broker {
class RecoveredDequeue : public TxOp{
Queue::shared_ptr queue;
- Message::shared_ptr msg;
+ intrusive_ptr<Message> msg;
public:
- RecoveredDequeue(Queue::shared_ptr queue, Message::shared_ptr msg);
+ RecoveredDequeue(Queue::shared_ptr queue, intrusive_ptr<Message> msg);
virtual bool prepare(TransactionContext* ctxt) throw();
virtual void commit() throw();
virtual void rollback() throw();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.cpp Wed Nov 14 13:02:14 2007
@@ -22,7 +22,7 @@
using namespace qpid::broker;
-RecoveredEnqueue::RecoveredEnqueue(Queue::shared_ptr _queue, Message::shared_ptr _msg) : queue(_queue), msg(_msg) {}
+RecoveredEnqueue::RecoveredEnqueue(Queue::shared_ptr _queue, intrusive_ptr<Message> _msg) : queue(_queue), msg(_msg) {}
bool RecoveredEnqueue::prepare(TransactionContext*) throw(){
//should never be called; transaction has already prepared if an enqueue is recovered
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveredEnqueue.h Wed Nov 14 13:02:14 2007
@@ -34,10 +34,10 @@
namespace broker {
class RecoveredEnqueue : public TxOp{
Queue::shared_ptr queue;
- Message::shared_ptr msg;
+ intrusive_ptr<Message> msg;
public:
- RecoveredEnqueue(Queue::shared_ptr queue, Message::shared_ptr msg);
+ RecoveredEnqueue(Queue::shared_ptr queue, intrusive_ptr<Message> msg);
virtual bool prepare(TransactionContext* ctxt) throw();
virtual void commit() throw();
virtual void rollback() throw();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp Wed Nov 14 13:02:14 2007
@@ -41,7 +41,7 @@
class RecoverableMessageImpl : public RecoverableMessage
{
- Message::shared_ptr msg;
+ intrusive_ptr<Message> msg;
const uint64_t stagingThreshold;
public:
RecoverableMessageImpl(Message::shared_ptr& _msg, uint64_t _stagingThreshold)
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp Wed Nov 14 13:02:14 2007
@@ -146,7 +146,7 @@
void SemanticHandler::handleContent(AMQFrame& frame)
{
- Message::shared_ptr msg(msgBuilder.getMessage());
+ intrusive_ptr<Message> msg(msgBuilder.getMessage());
if (!msg) {//start of frameset will be indicated by frame flags
msgBuilder.start(incoming.next());
msg = msgBuilder.getMessage();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp Wed Nov 14 13:02:14 2007
@@ -230,7 +230,7 @@
delivery.addTo(outstanding);
}
-bool SemanticState::checkPrefetch(Message::shared_ptr& msg)
+bool SemanticState::checkPrefetch(intrusive_ptr<Message>& msg)
{
Mutex::ScopedLock locker(deliveryLock);
bool countOk = !prefetchCount || prefetchCount > unacked.size();
@@ -288,13 +288,13 @@
}
}
-bool SemanticState::ConsumerImpl::filter(Message::shared_ptr msg)
+bool SemanticState::ConsumerImpl::filter(intrusive_ptr<Message> msg)
{
return !(nolocal &&
&parent->getSession().getConnection() == msg->getPublisher());
}
-bool SemanticState::ConsumerImpl::checkCredit(Message::shared_ptr& msg)
+bool SemanticState::ConsumerImpl::checkCredit(intrusive_ptr<Message>& msg)
{
Mutex::ScopedLock l(lock);
if (msgCredit == 0 || (byteCredit != 0xFFFFFFFF && byteCredit < msg->getRequiredCredit())) {
@@ -331,7 +331,7 @@
}
}
-void SemanticState::handle(Message::shared_ptr msg) {
+void SemanticState::handle(intrusive_ptr<Message> msg) {
if (txBuffer.get()) {
TxPublish* deliverable(new TxPublish(msg));
TxOp::shared_ptr op(deliverable);
@@ -343,7 +343,7 @@
}
}
-void SemanticState::route(Message::shared_ptr msg, Deliverable& strategy) {
+void SemanticState::route(intrusive_ptr<Message> msg, Deliverable& strategy) {
std::string exchangeName = msg->getExchangeName();
if (!cacheExchange || cacheExchange->getName() != exchangeName){
cacheExchange = session.getConnection().broker.getExchanges().get(exchangeName);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.h Wed Nov 14 13:02:14 2007
@@ -68,7 +68,7 @@
uint32_t msgCredit;
uint32_t byteCredit;
- bool checkCredit(Message::shared_ptr& msg);
+ bool checkCredit(intrusive_ptr<Message>& msg);
public:
typedef shared_ptr<ConsumerImpl> shared_ptr;
@@ -78,7 +78,7 @@
bool ack, bool nolocal, bool acquire);
~ConsumerImpl();
bool deliver(QueuedMessage& msg);
- bool filter(Message::shared_ptr msg);
+ bool filter(intrusive_ptr<Message> msg);
void setWindowMode();
void setCreditMode();
@@ -124,9 +124,9 @@
boost::shared_ptr<Exchange> cacheExchange;
- void route(Message::shared_ptr msg, Deliverable& strategy);
+ void route(intrusive_ptr<Message> msg, Deliverable& strategy);
void record(const DeliveryRecord& delivery);
- bool checkPrefetch(Message::shared_ptr& msg);
+ bool checkPrefetch(intrusive_ptr<Message>& msg);
void checkDtxTimeout();
ConsumerImpl::shared_ptr find(const std::string& destination);
void ack(DeliveryId deliveryTag, DeliveryId endTag, bool cumulative);
@@ -187,7 +187,7 @@
void acquire(DeliveryId first, DeliveryId last, std::vector<DeliveryId>& acquired);
void release(DeliveryId first, DeliveryId last);
void reject(DeliveryId first, DeliveryId last);
- void handle(Message::shared_ptr msg);
+ void handle(intrusive_ptr<Message> msg);
};
}} // namespace qpid::broker
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.cpp Wed Nov 14 13:02:14 2007
@@ -23,7 +23,7 @@
using namespace qpid::broker;
-TxPublish::TxPublish(Message::shared_ptr _msg) : msg(_msg) {}
+TxPublish::TxPublish(intrusive_ptr<Message> _msg) : msg(_msg) {}
bool TxPublish::prepare(TransactionContext* ctxt) throw(){
try{
@@ -47,7 +47,7 @@
delivered = true;
}
-TxPublish::Prepare::Prepare(TransactionContext* _ctxt, Message::shared_ptr& _msg)
+TxPublish::Prepare::Prepare(TransactionContext* _ctxt, intrusive_ptr<Message>& _msg)
: ctxt(_ctxt), msg(_msg){}
void TxPublish::Prepare::operator()(Queue::shared_ptr& queue){
@@ -61,7 +61,7 @@
}
}
-TxPublish::Commit::Commit(Message::shared_ptr& _msg) : msg(_msg){}
+TxPublish::Commit::Commit(intrusive_ptr<Message>& _msg) : msg(_msg){}
void TxPublish::Commit::operator()(Queue::shared_ptr& queue){
queue->process(msg);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/TxPublish.h Wed Nov 14 13:02:14 2007
@@ -45,24 +45,24 @@
class TxPublish : public TxOp, public Deliverable{
class Prepare{
TransactionContext* ctxt;
- Message::shared_ptr& msg;
+ intrusive_ptr<Message>& msg;
public:
- Prepare(TransactionContext* ctxt, Message::shared_ptr& msg);
+ Prepare(TransactionContext* ctxt, intrusive_ptr<Message>& msg);
void operator()(Queue::shared_ptr& queue);
};
class Commit{
- Message::shared_ptr& msg;
+ intrusive_ptr<Message>& msg;
public:
- Commit(Message::shared_ptr& msg);
+ Commit(intrusive_ptr<Message>& msg);
void operator()(Queue::shared_ptr& queue);
};
- Message::shared_ptr msg;
+ intrusive_ptr<Message> msg;
std::list<Queue::shared_ptr> queues;
public:
- TxPublish(Message::shared_ptr msg);
+ TxPublish(intrusive_ptr<Message> msg);
virtual bool prepare(TransactionContext* ctxt) throw();
virtual void commit() throw();
virtual void rollback() throw();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/SessionManager.cpp Wed Nov 14 13:02:14 2007
@@ -74,8 +74,8 @@
virtual void send(const AMQBody&) {}
//delivery adapter methods, also no-ops:
- virtual DeliveryId deliver(Message::shared_ptr&, DeliveryToken::shared_ptr) { return 0; }
- virtual void redeliver(Message::shared_ptr&, DeliveryToken::shared_ptr, DeliveryId) {}
+ virtual DeliveryId deliver(intrusive_ptr<Message>&, DeliveryToken::shared_ptr) { return 0; }
+ virtual void redeliver(intrusive_ptr<Message>&, DeliveryToken::shared_ptr, DeliveryId) {}
};
SessionManager::~SessionManager(){}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementAgent.cpp Wed Nov 14 13:02:14 2007
@@ -96,7 +96,7 @@
if (managementObjects.empty ())
return;
- Message::shared_ptr msg (new Message ());
+ intrusive_ptr<Message> msg (new Message ());
// Build the magic number for the management message.
msgBuffer.putOctet ('A');
@@ -294,7 +294,7 @@
iter->second->doMethod (methodName, inBuffer, outBuffer);
}
- Message::shared_ptr outMsg (new Message ());
+ intrusive_ptr<Message> outMsg (new Message ());
uint32_t msgSize = 4096 - outBuffer.available ();
outBuffer.reset ();
AMQFrame method (0, MessageTransferBody(ProtocolVersion(),
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp Wed Nov 14 13:02:14 2007
@@ -59,16 +59,16 @@
struct DeliveryRecorder : DeliveryAdapter
{
DeliveryId id;
- typedef std::pair<Message::shared_ptr, DeliveryToken::shared_ptr> Delivery;
+ typedef std::pair<intrusive_ptr<Message>, DeliveryToken::shared_ptr> Delivery;
std::vector<Delivery> delivered;
- DeliveryId deliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token)
+ DeliveryId deliver(intrusive_ptr<Message>& msg, DeliveryToken::shared_ptr token)
{
delivered.push_back(Delivery(msg, token));
return ++id;
}
- void redeliver(Message::shared_ptr& msg, DeliveryToken::shared_ptr token, DeliveryId /*tag*/)
+ void redeliver(intrusive_ptr<Message>& msg, DeliveryToken::shared_ptr token, DeliveryId /*tag*/)
{
delivered.push_back(Delivery(msg, token));
}
@@ -215,7 +215,7 @@
void testDeliveryNoAck(){
Channel channel(connection, recorder, 7);
- Message::shared_ptr msg(createMessage("test", "my_routing_key", "my_message_id", 14));
+ intrusive_ptr<Message> msg(createMessage("test", "my_routing_key", "my_message_id", 14));
Queue::shared_ptr queue(new Queue("my_queue"));
string tag("test");
DeliveryToken::shared_ptr token(MessageDelivery::getBasicConsumeToken("my-token"));
@@ -239,9 +239,9 @@
const string data1("abcd");
const string data2("efghijk");
const string data3("lmnopqrstuvwxyz");
- Message::shared_ptr msg1(createMessage("e", "A", "MsgA", data1.size()));
- Message::shared_ptr msg2(createMessage("e", "B", "MsgB", data2.size()));
- Message::shared_ptr msg3(createMessage("e", "C", "MsgC", data3.size()));
+ intrusive_ptr<Message> msg1(createMessage("e", "A", "MsgA", data1.size()));
+ intrusive_ptr<Message> msg2(createMessage("e", "B", "MsgB", data2.size()));
+ intrusive_ptr<Message> msg3(createMessage("e", "C", "MsgC", data3.size()));
addContent(msg1, data1);
addContent(msg2, data2);
addContent(msg3, data3);
@@ -261,7 +261,7 @@
queue->deliver(msg3);
sleep(2);
- Message::shared_ptr next = queue->dequeue().payload;
+ intrusive_ptr<Message> next = queue->dequeue().payload;
CPPUNIT_ASSERT_EQUAL(msg1, next);
CPPUNIT_ASSERT_EQUAL((uint32_t) data1.size(), next->encodedContentSize());
next = queue->dequeue().payload;
@@ -289,7 +289,7 @@
MockMessageStore store;
{//must ensure that store is last thing deleted
const string data1("abcd");
- Message::shared_ptr msg1(createMessage("e", "A", "MsgA", data1.size()));
+ intrusive_ptr<Message> msg1(createMessage("e", "A", "MsgA", data1.size()));
addContent(msg1, data1);
Queue::shared_ptr queue1(new Queue("my_queue1", false, &store, 0));
@@ -300,7 +300,7 @@
queue3->deliver(msg1);
sleep(2);
- Message::shared_ptr next = queue1->dequeue().payload;
+ intrusive_ptr<Message> next = queue1->dequeue().payload;
CPPUNIT_ASSERT_EQUAL(msg1, next);
next = queue2->dequeue().payload;
CPPUNIT_ASSERT_EQUAL(msg1, next);
@@ -327,7 +327,7 @@
channel.flow(false);
//'publish' a message
- Message::shared_ptr msg(createMessage("test", "my_routing_key", "my_message_id", 14));
+ intrusive_ptr<Message> msg(createMessage("test", "my_routing_key", "my_message_id", 14));
addContent(msg, "abcdefghijklmn");
queue->deliver(msg);
@@ -342,9 +342,9 @@
CPPUNIT_ASSERT_EQUAL(token, recorder.delivered.front().second);
}
- Message::shared_ptr createMessage(const string& exchange, const string& routingKey, const string& messageId, uint64_t contentSize)
+ intrusive_ptr<Message> createMessage(const string& exchange, const string& routingKey, const string& messageId, uint64_t contentSize)
{
- Message::shared_ptr msg(new Message());
+ intrusive_ptr<Message> msg(new Message());
AMQFrame method(0, MessageTransferBody(ProtocolVersion(), 0, exchange, 0, 0));
AMQFrame header(0, AMQHeaderBody());
@@ -358,7 +358,7 @@
return msg;
}
- void addContent(Message::shared_ptr msg, const string& data)
+ void addContent(intrusive_ptr<Message> msg, const string& data)
{
AMQFrame content(0, AMQContentBody(data));
msg->getFrames().append(content);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/ExchangeTest.cpp Wed Nov 14 13:02:14 2007
@@ -64,7 +64,7 @@
queue.reset();
queue2.reset();
- Message::shared_ptr msgPtr(MessageUtils::createMessage("exchange", "key", "id"));
+ intrusive_ptr<Message> msgPtr(MessageUtils::createMessage("exchange", "key", "id"));
DeliverableMessage msg(msgPtr);
topic.route(msg, "abc", 0);
direct.route(msg, "abc", 0);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp Wed Nov 14 13:02:14 2007
@@ -49,7 +49,7 @@
string data1("abcdefg");
string data2("hijklmn");
- Message::shared_ptr msg(new Message());
+ intrusive_ptr<Message> msg(new Message());
AMQFrame method(0, MessageTransferBody(ProtocolVersion(), 0, exchange, 0, 0));
AMQFrame header(0, AMQHeaderBody());
@@ -77,7 +77,7 @@
msg->encode(wbuffer);
Buffer rbuffer(buff, msg->encodedSize());
- msg.reset(new Message());
+ msg = new Message();
msg->decodeHeader(rbuffer);
msg->decodeContent(rbuffer);
CPPUNIT_ASSERT_EQUAL(exchange, msg->getExchangeName());
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/MessageUtils.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/MessageUtils.h?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/MessageUtils.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/MessageUtils.h Wed Nov 14 13:02:14 2007
@@ -23,15 +23,16 @@
#include "qpid/broker/MessageDelivery.h"
#include "qpid/framing/AMQFrame.h"
-using namespace qpid::broker;
-using namespace qpid::framing;
+using namespace qpid;
+using namespace broker;
+using namespace framing;
struct MessageUtils
{
- static Message::shared_ptr createMessage(const string& exchange, const string& routingKey,
+ static intrusive_ptr<Message> createMessage(const string& exchange, const string& routingKey,
const string& messageId, uint64_t contentSize = 0)
{
- Message::shared_ptr msg(new Message());
+ intrusive_ptr<Message> msg(new Message());
AMQFrame method(0, MessageTransferBody(ProtocolVersion(), 0, exchange, 0, 0));
AMQFrame header(0, AMQHeaderBody());
@@ -45,7 +46,7 @@
return msg;
}
- static void addContent(Message::shared_ptr msg, const string& data)
+ static void addContent(intrusive_ptr<Message> msg, const string& data)
{
AMQFrame content(0, AMQContentBody(data));
msg->getFrames().append(content);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp Wed Nov 14 13:02:14 2007
@@ -38,7 +38,7 @@
public:
typedef shared_ptr<TestConsumer> shared_ptr;
- Message::shared_ptr last;
+ intrusive_ptr<Message> last;
bool received;
TestConsumer(): received(false) {};
@@ -71,8 +71,8 @@
public:
- Message::shared_ptr message(std::string exchange, std::string routingKey) {
- Message::shared_ptr msg(new Message());
+ intrusive_ptr<Message> message(std::string exchange, std::string routingKey) {
+ intrusive_ptr<Message> msg(new Message());
AMQFrame method(0, MessageTransferBody(ProtocolVersion(), 0, exchange, 0, 0));
AMQFrame header(0, AMQHeaderBody());
msg->getFrames().append(method);
@@ -85,14 +85,14 @@
void testAsyncMessage(){
Queue::shared_ptr queue(new Queue("my_test_queue", true));
- Message::shared_ptr received;
+ intrusive_ptr<Message> received;
TestConsumer::shared_ptr c1(new TestConsumer());
queue->consume(c1);
//Test basic delivery:
- Message::shared_ptr msg1 = message("e", "A");
+ intrusive_ptr<Message> msg1 = message("e", "A");
msg1->enqueueAsync();//this is done on enqueue which is not called from process
queue->process(msg1);
sleep(2);
@@ -109,7 +109,7 @@
void testAsyncMessageCount(){
Queue::shared_ptr queue(new Queue("my_test_queue", true));
- Message::shared_ptr msg1 = message("e", "A");
+ intrusive_ptr<Message> msg1 = message("e", "A");
msg1->enqueueAsync();//this is done on enqueue which is not called from process
queue->process(msg1);
@@ -134,9 +134,9 @@
CPPUNIT_ASSERT_EQUAL(uint32_t(2), queue->getConsumerCount());
//Test basic delivery:
- Message::shared_ptr msg1 = message("e", "A");
- Message::shared_ptr msg2 = message("e", "B");
- Message::shared_ptr msg3 = message("e", "C");
+ intrusive_ptr<Message> msg1 = message("e", "A");
+ intrusive_ptr<Message> msg2 = message("e", "B");
+ intrusive_ptr<Message> msg3 = message("e", "C");
queue->deliver(msg1);
if (!c1->received)
@@ -183,10 +183,10 @@
void testDequeue(){
Queue::shared_ptr queue(new Queue("my_queue", true));
- Message::shared_ptr msg1 = message("e", "A");
- Message::shared_ptr msg2 = message("e", "B");
- Message::shared_ptr msg3 = message("e", "C");
- Message::shared_ptr received;
+ intrusive_ptr<Message> msg1 = message("e", "A");
+ intrusive_ptr<Message> msg2 = message("e", "B");
+ intrusive_ptr<Message> msg3 = message("e", "C");
+ intrusive_ptr<Message> received;
queue->deliver(msg1);
queue->deliver(msg2);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/TxAckTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/TxAckTest.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/TxAckTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/TxAckTest.cpp Wed Nov 14 13:02:14 2007
@@ -29,6 +29,7 @@
using std::list;
using std::vector;
+using namespace qpid;
using namespace qpid::broker;
using namespace qpid::framing;
@@ -58,7 +59,7 @@
AccumulatedAck acked;
TestMessageStore store;
Queue::shared_ptr queue;
- vector<Message::shared_ptr> messages;
+ vector<intrusive_ptr<Message> > messages;
list<DeliveryRecord> deliveries;
TxAck op;
@@ -68,7 +69,7 @@
TxAckTest() : acked(0), queue(new Queue("my_queue", false, &store, 0)), op(acked, deliveries)
{
for(int i = 0; i < 10; i++){
- Message::shared_ptr msg(new Message());
+ intrusive_ptr<Message> msg(new Message());
AMQFrame method(0, MessageTransferBody(ProtocolVersion(), 0, "exchange", 0, 0));
AMQFrame header(0, AMQHeaderBody());
msg->getFrames().append(method);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/TxPublishTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/TxPublishTest.cpp?rev=595056&r1=595055&r2=595056&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/TxPublishTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/TxPublishTest.cpp Wed Nov 14 13:02:14 2007
@@ -63,7 +63,7 @@
TestMessageStore store;
Queue::shared_ptr queue1;
Queue::shared_ptr queue2;
- Message::shared_ptr msg;
+ intrusive_ptr<Message> msg;
TxPublish op;
public:
@@ -99,7 +99,7 @@
op.prepare(0);
op.commit();
CPPUNIT_ASSERT_EQUAL((uint32_t) 1, queue1->getMessageCount());
- Message::shared_ptr msg_dequeue = queue1->dequeue().payload;
+ intrusive_ptr<Message> msg_dequeue = queue1->dequeue().payload;
CPPUNIT_ASSERT_EQUAL( true, ((PersistableMessage*) msg_dequeue.get())->isEnqueueComplete());
CPPUNIT_ASSERT_EQUAL(msg, msg_dequeue);