* Further (minor) changes to the interface between store and broker.

* TxBuffer now uses shared_ptr to TxOps (removed DeletingTxOp)
* Queue now persists the field table of settings 



git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@525801 13f79535-47bb-0310-9956-ffa450edef68

16 files changed, 137 insertions, 170 deletions

diff --git a/qpid/cpp/src/Makefile.am b/qpid/cpp/src/Makefile.am
index a63c68111a..00e31a7d1a 100644
--- a/qpid/cpp/src/Makefile.am
+++ b/qpid/cpp/src/Makefile.am
@@ -136,8 +136,6 @@ libqpidbroker_la_SOURCES =			\
   $(broker)/ConnectionToken.h				\
   $(broker)/Consumer.h					\
   $(broker)/Content.h					\
-  $(broker)/DeletingTxOp.cpp				\
-  $(broker)/DeletingTxOp.h				\
   $(broker)/Deliverable.h					\
   $(broker)/DeliverableMessage.cpp			\
   $(broker)/DeliverableMessage.h				\
diff --git a/qpid/cpp/src/broker/BrokerChannel.cpp b/qpid/cpp/src/broker/BrokerChannel.cpp
index 5f35f36da6..bd97c6f021 100644
--- a/qpid/cpp/src/broker/BrokerChannel.cpp
+++ b/qpid/cpp/src/broker/BrokerChannel.cpp
@@ -28,7 +28,6 @@
 #include <boost/bind.hpp>
 
 #include "BrokerChannel.h"
-#include "DeletingTxOp.h"
 #include "../framing/ChannelAdapter.h"
 #include "../QpidError.h"
 #include "DeliverableMessage.h"
@@ -110,8 +109,8 @@ void Channel::begin(){
 }
 
 void Channel::commit(){
-    TxAck txAck(accumulatedAck, unacked);
-    txBuffer.enlist(&txAck);
+    TxOp::shared_ptr txAck(new TxAck(accumulatedAck, unacked));
+    txBuffer.enlist(txAck);
     if(txBuffer.prepare(store)){
         txBuffer.commit();
     }
@@ -186,11 +185,12 @@ void Channel::handleInlineTransfer(Message::shared_ptr msg)
     Exchange::shared_ptr exchange =
         connection.broker.getExchanges().get(msg->getExchange());
     if(transactional){
-        TxPublish* deliverable = new TxPublish(msg);
+        TxPublish* deliverable(new TxPublish(msg));
+        TxOp::shared_ptr op(deliverable);
         exchange->route(
             *deliverable, msg->getRoutingKey(),
             &(msg->getApplicationHeaders()));
-        txBuffer.enlist(new DeletingTxOp(deliverable));
+        txBuffer.enlist(op);
     }else{
         DeliverableMessage deliverable(msg);
         exchange->route(
@@ -223,10 +223,11 @@ void Channel::complete(Message::shared_ptr msg) {
         connection.broker.getExchanges().get(msg->getExchange());
     assert(exchange.get());
     if(transactional) {
-        std::auto_ptr<TxPublish> deliverable(new TxPublish(msg));
+        TxPublish* deliverable(new TxPublish(msg));
+        TxOp::shared_ptr op(deliverable);
         exchange->route(*deliverable, msg->getRoutingKey(),
                         &(msg->getApplicationHeaders()));
-        txBuffer.enlist(new DeletingTxOp(deliverable.release()));
+        txBuffer.enlist(op);
     } else {
         DeliverableMessage deliverable(msg);
         exchange->route(deliverable, msg->getRoutingKey(),
diff --git a/qpid/cpp/src/broker/BrokerMessageBase.h b/qpid/cpp/src/broker/BrokerMessageBase.h
index 93178774dd..41f54bc8bb 100644
--- a/qpid/cpp/src/broker/BrokerMessageBase.h
+++ b/qpid/cpp/src/broker/BrokerMessageBase.h
@@ -89,7 +89,7 @@ class Message : public PersistableMessage{
     
     void setRouting(const std::string& _exchange, const std::string& _routingKey)
     { exchange = _exchange; routingKey = _routingKey; } 
-    void setPersistenceId(uint64_t _persistenceId) { persistenceId = _persistenceId; } // XXXX: Only used in tests?
+    void setPersistenceId(uint64_t _persistenceId) const { persistenceId = _persistenceId; }
     void redeliver() { redelivered = true; }
 
     /**
@@ -171,7 +171,7 @@ class Message : public PersistableMessage{
     std::string routingKey;
     const bool mandatory;
     const bool immediate;
-    uint64_t persistenceId;
+    mutable uint64_t persistenceId;
     bool redelivered;
     AMQMethodBodyPtr respondTo;
 };
diff --git a/qpid/cpp/src/broker/BrokerQueue.cpp b/qpid/cpp/src/broker/BrokerQueue.cpp
index 08288a9dbd..5d34a52fcc 100644
--- a/qpid/cpp/src/broker/BrokerQueue.cpp
+++ b/qpid/cpp/src/broker/BrokerQueue.cpp
@@ -216,19 +216,20 @@ namespace
     const std::string qpidMaxCount("qpid.max_count");
 }
 
-void Queue::create(const FieldTable& settings)
+void Queue::create(const FieldTable& _settings)
 {
+    settings = _settings;
     //TODO: hold onto settings and persist them as part of encode
     //      in fact settings should be passed in on construction
     if (store) {
         store->create(*this);
     }
-    configure(settings);
+    configure(_settings);
 }
 
-void Queue::configure(const FieldTable& settings)
+void Queue::configure(const FieldTable& _settings)
 {
-    std::auto_ptr<QueuePolicy> _policy(new QueuePolicy(settings));
+    std::auto_ptr<QueuePolicy> _policy(new QueuePolicy(_settings));
     if (_policy->getMaxCount() || _policy->getMaxSize()) 
         setPolicy(_policy);
 }
@@ -255,7 +256,7 @@ uint64_t Queue::getPersistenceId() const
     return persistenceId; 
 }
 
-void Queue::setPersistenceId(uint64_t _persistenceId)
+void Queue::setPersistenceId(uint64_t _persistenceId) const
 { 
     persistenceId = _persistenceId; 
 }
@@ -263,13 +264,12 @@ void Queue::setPersistenceId(uint64_t _persistenceId)
 void Queue::encode(framing::Buffer& buffer) const 
 {
     buffer.putShortString(name);
-    //TODO store all required properties
+    buffer.putFieldTable(settings);
 }
 
 uint32_t Queue::encodedSize() const
 {
-    //TODO, revise when storing full set of queue properties
-    return name.size() + 1/*short string size octet*/;
+    return name.size() + 1/*short string size octet*/ + settings.size();
 }
 
 Queue::shared_ptr Queue::decode(QueueRegistry& queues, framing::Buffer& buffer)
@@ -277,6 +277,8 @@ Queue::shared_ptr Queue::decode(QueueRegistry& queues, framing::Buffer& buffer)
     string name;
     buffer.getShortString(name);
     std::pair<Queue::shared_ptr, bool> result = queues.declare(name, true);
+    buffer.getFieldTable(result.first->settings);
+    result.first->configure(result.first->settings);
     return result.first;
 }
 
diff --git a/qpid/cpp/src/broker/BrokerQueue.h b/qpid/cpp/src/broker/BrokerQueue.h
index 009f95b384..808a523869 100644
--- a/qpid/cpp/src/broker/BrokerQueue.h
+++ b/qpid/cpp/src/broker/BrokerQueue.h
@@ -70,6 +70,7 @@ namespace qpid {
             int64_t lastUsed;
             Consumer* exclusive;
             mutable uint64_t persistenceId;
+            framing::FieldTable settings;
             std::auto_ptr<QueuePolicy> policy;            
 
             void pop();
@@ -138,7 +139,7 @@ namespace qpid {
 
             //PersistableQueue support:
             uint64_t getPersistenceId() const;
-            void setPersistenceId(uint64_t persistenceId);
+            void setPersistenceId(uint64_t persistenceId) const;
             void encode(framing::Buffer& buffer) const;
             uint32_t encodedSize() const;
 
diff --git a/qpid/cpp/src/broker/DeletingTxOp.cpp b/qpid/cpp/src/broker/DeletingTxOp.cpp
deleted file mode 100644
index ae6ac0f2e6..0000000000
--- a/qpid/cpp/src/broker/DeletingTxOp.cpp
+++ /dev/null
@@ -1,45 +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 "DeletingTxOp.h"
-
-using namespace qpid::broker;
-
-DeletingTxOp::DeletingTxOp(TxOp* const _delegate) : delegate(_delegate){}
-
-bool DeletingTxOp::prepare(TransactionContext* ctxt) throw(){
-    return delegate && delegate->prepare(ctxt);
-}
-
-void DeletingTxOp::commit() throw(){
-    if(delegate){
-        delegate->commit();
-        delete delegate;
-        delegate = 0;
-    }
-}
-
-void DeletingTxOp::rollback() throw(){
-    if(delegate){
-        delegate->rollback();
-        delete delegate;
-        delegate = 0;
-    }
-}
diff --git a/qpid/cpp/src/broker/DeletingTxOp.h b/qpid/cpp/src/broker/DeletingTxOp.h
deleted file mode 100644
index ddef1c77c0..0000000000
--- a/qpid/cpp/src/broker/DeletingTxOp.h
+++ /dev/null
@@ -1,45 +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.
- *
- */
-#ifndef _DeletingTxOp_
-#define _DeletingTxOp_
-
-#include "TxOp.h"
-
-namespace qpid {
-    namespace broker {
-        /**
-         * TxOp wrapper that will delegate calls & delete the object
-         * to which it delegates after completion of the transaction.
-         */
-        class DeletingTxOp : public virtual TxOp{
-            TxOp* delegate;
-        public:
-            DeletingTxOp(TxOp* const delegate);
-            virtual bool prepare(TransactionContext* ctxt) throw();
-            virtual void commit()  throw();
-            virtual void rollback()  throw();
-            virtual ~DeletingTxOp(){}
-        };
-    }
-}
-
-
-#endif
diff --git a/qpid/cpp/src/broker/Persistable.h b/qpid/cpp/src/broker/Persistable.h
index a53ea428ed..58ea1a7322 100644
--- a/qpid/cpp/src/broker/Persistable.h
+++ b/qpid/cpp/src/broker/Persistable.h
@@ -37,7 +37,7 @@ public:
     /**
      * Allows the store to attach its own identifier to this object
      */
-    virtual void setPersistenceId(uint64_t id) = 0;
+    virtual void setPersistenceId(uint64_t id) const = 0;
     /**
      * Returns any identifier the store may have attached to this
      * object
diff --git a/qpid/cpp/src/broker/RecoverableMessage.h b/qpid/cpp/src/broker/RecoverableMessage.h
index c79a84081c..3dd892d577 100644
--- a/qpid/cpp/src/broker/RecoverableMessage.h
+++ b/qpid/cpp/src/broker/RecoverableMessage.h
@@ -36,6 +36,7 @@ class RecoverableMessage
 {
 public:
     typedef boost::shared_ptr<RecoverableMessage> shared_ptr;
+    virtual void setPersistenceId(uint64_t id) = 0;
     /**
      * Used by store to determine whether to load content on recovery
      * or let message load its own content as and when it requires it.
diff --git a/qpid/cpp/src/broker/RecoverableQueue.h b/qpid/cpp/src/broker/RecoverableQueue.h
index a5c564b947..656a5f0149 100644
--- a/qpid/cpp/src/broker/RecoverableQueue.h
+++ b/qpid/cpp/src/broker/RecoverableQueue.h
@@ -36,6 +36,8 @@ class RecoverableQueue
 {
 public:
     typedef boost::shared_ptr<RecoverableQueue> shared_ptr;
+
+    virtual void setPersistenceId(uint64_t id) = 0;
     /**
      * Used during recovery to add stored messages back to the queue
      */
diff --git a/qpid/cpp/src/broker/RecoveryManagerImpl.cpp b/qpid/cpp/src/broker/RecoveryManagerImpl.cpp
index 7c04ec9916..f82399f95c 100644
--- a/qpid/cpp/src/broker/RecoveryManagerImpl.cpp
+++ b/qpid/cpp/src/broker/RecoveryManagerImpl.cpp
@@ -45,6 +45,7 @@ public:
     RecoverableMessageImpl(Message::shared_ptr& _msg, uint64_t _stagingThreshold) 
         : msg(_msg), stagingThreshold(_stagingThreshold) {}
     ~RecoverableMessageImpl() {};
+    void setPersistenceId(uint64_t id);
     bool loadContent(uint64_t available);
     void decodeContent(framing::Buffer& buffer);
     void recover(Queue::shared_ptr queue);
@@ -56,6 +57,7 @@ class RecoverableQueueImpl : public RecoverableQueue
 public:
     RecoverableQueueImpl(Queue::shared_ptr& _queue) : queue(_queue) {}
     ~RecoverableQueueImpl() {};
+    void setPersistenceId(uint64_t id);
     void recover(RecoverableMessage::shared_ptr msg);
 };
 
@@ -125,7 +127,17 @@ void RecoverableMessageImpl::recover(Queue::shared_ptr queue)
     queue->recover(msg);
 }
 
+void RecoverableMessageImpl::setPersistenceId(uint64_t id)
+{
+    msg->setPersistenceId(id);
+}
+
 void RecoverableQueueImpl::recover(RecoverableMessage::shared_ptr msg)
 {
     dynamic_pointer_cast<RecoverableMessageImpl>(msg)->recover(queue);
 }
+
+void RecoverableQueueImpl::setPersistenceId(uint64_t id)
+{
+    queue->setPersistenceId(id);
+}
diff --git a/qpid/cpp/src/broker/TxBuffer.cpp b/qpid/cpp/src/broker/TxBuffer.cpp
index 47bc13b5c0..c526c88c25 100644
--- a/qpid/cpp/src/broker/TxBuffer.cpp
+++ b/qpid/cpp/src/broker/TxBuffer.cpp
@@ -20,7 +20,8 @@
  */
 #include "TxBuffer.h"
 
-using std::mem_fun;
+#include <boost/mem_fn.hpp>
+using boost::mem_fn;
 using namespace qpid::broker;
 
 bool TxBuffer::prepare(TransactionalStore* const store)
@@ -39,17 +40,17 @@ bool TxBuffer::prepare(TransactionalStore* const store)
 
 void TxBuffer::commit()
 {
-    for_each(ops.begin(), ops.end(), mem_fun(&TxOp::commit));
+    for_each(ops.begin(), ops.end(), mem_fn(&TxOp::commit));
     ops.clear();
 }
 
 void TxBuffer::rollback()
 {
-    for_each(ops.begin(), ops.end(), mem_fun(&TxOp::rollback));
+    for_each(ops.begin(), ops.end(), mem_fn(&TxOp::rollback));
     ops.clear();
 }
 
-void TxBuffer::enlist(TxOp* const op)
+void TxBuffer::enlist(TxOp::shared_ptr op)
 {
     ops.push_back(op);
 }
diff --git a/qpid/cpp/src/broker/TxBuffer.h b/qpid/cpp/src/broker/TxBuffer.h
index 61c0820c7d..86ee8945b5 100644
--- a/qpid/cpp/src/broker/TxBuffer.h
+++ b/qpid/cpp/src/broker/TxBuffer.h
@@ -59,8 +59,8 @@
 namespace qpid {
     namespace broker {
         class TxBuffer{
-            typedef std::vector<TxOp*>::iterator op_iterator;
-            std::vector<TxOp*> ops;
+            typedef std::vector<TxOp::shared_ptr>::iterator op_iterator;
+            std::vector<TxOp::shared_ptr> ops;
         public:
             /**
              * Requests that all ops are prepared. This should
@@ -98,7 +98,7 @@ namespace qpid {
             /**
              * Adds an operation to the transaction.
              */
-            void enlist(TxOp* const op);
+            void enlist(TxOp::shared_ptr op);
         };
     }
 }
diff --git a/qpid/cpp/src/broker/TxOp.h b/qpid/cpp/src/broker/TxOp.h
index 8546a42616..e687c437cc 100644
--- a/qpid/cpp/src/broker/TxOp.h
+++ b/qpid/cpp/src/broker/TxOp.h
@@ -22,11 +22,14 @@
 #define _TxOp_
 
 #include "TransactionalStore.h"
+#include <boost/shared_ptr.hpp>
 
 namespace qpid {
     namespace broker {
         class TxOp{
         public:
+            typedef boost::shared_ptr<TxOp> shared_ptr;
+
             virtual bool prepare(TransactionContext*) throw() = 0;
             virtual void commit()  throw() = 0;
             virtual void rollback()  throw() = 0;
diff --git a/qpid/cpp/src/tests/FieldTableTest.cpp b/qpid/cpp/src/tests/FieldTableTest.cpp
index f485ca187e..a7a03cc574 100644
--- a/qpid/cpp/src/tests/FieldTableTest.cpp
+++ b/qpid/cpp/src/tests/FieldTableTest.cpp
@@ -28,6 +28,7 @@ class FieldTableTest : public CppUnit::TestCase
 {
     CPPUNIT_TEST_SUITE(FieldTableTest);
     CPPUNIT_TEST(testMe);
+    CPPUNIT_TEST(testAssignment);
     CPPUNIT_TEST_SUITE_END();
 
   public:
@@ -46,6 +47,38 @@ class FieldTableTest : public CppUnit::TestCase
         CPPUNIT_ASSERT_EQUAL(std::string("BCDE"), ft2.getString("A"));
 
     }
+
+    void testAssignment()
+    {
+        FieldTable a;
+        FieldTable b;
+
+        a.setString("A", "BBBB");
+        a.setInt("B", 1234);
+        b = a;
+        a.setString("A", "CCCC");
+        
+        CPPUNIT_ASSERT_EQUAL(std::string("CCCC"), a.getString("A"));
+        CPPUNIT_ASSERT_EQUAL(std::string("BBBB"), b.getString("A"));
+        CPPUNIT_ASSERT_EQUAL(1234, a.getInt("B"));
+        CPPUNIT_ASSERT_EQUAL(1234, b.getInt("B"));
+
+        FieldTable d;
+        {
+            FieldTable c;
+            c = a;
+            
+            Buffer buffer(c.size());
+            buffer.putFieldTable(c);
+            buffer.flip();     
+            buffer.getFieldTable(d);
+            CPPUNIT_ASSERT_EQUAL(c, d);
+            CPPUNIT_ASSERT_EQUAL(std::string("CCCC"), c.getString("A"));
+            CPPUNIT_ASSERT_EQUAL(1234, c.getInt("B"));
+        }
+        CPPUNIT_ASSERT_EQUAL(std::string("CCCC"), d.getString("A"));
+        CPPUNIT_ASSERT_EQUAL(1234, d.getInt("B"));
+    }
 };
 
 
diff --git a/qpid/cpp/src/tests/TxBufferTest.cpp b/qpid/cpp/src/tests/TxBufferTest.cpp
index 0d1fe7a04b..91ccc1e47d 100644
--- a/qpid/cpp/src/tests/TxBufferTest.cpp
+++ b/qpid/cpp/src/tests/TxBufferTest.cpp
@@ -24,6 +24,7 @@
 #include <vector>
 
 using namespace qpid::broker;
+using boost::static_pointer_cast;
 
 template <class T> void assertEqualVector(std::vector<T>& expected, std::vector<T>& actual){
     unsigned int i = 0;
@@ -43,6 +44,8 @@ class TxBufferTest : public CppUnit::TestCase
         std::vector<int> actual;
         bool failOnPrepare;
     public:
+        typedef boost::shared_ptr<MockTxOp> shared_ptr;
+
         MockTxOp() : failOnPrepare(false) {}
         MockTxOp(bool _failOnPrepare) : failOnPrepare(_failOnPrepare) {}
 
@@ -166,100 +169,100 @@ class TxBufferTest : public CppUnit::TestCase
         MockTransactionalStore store;
         store.expectBegin().expectCommit();
 
-        MockTxOp opA;
-        opA.expectPrepare().expectCommit();
-        MockTxOp opB;
-        opB.expectPrepare().expectPrepare().expectCommit().expectCommit();//opB enlisted twice to test reative order
-        MockTxOp opC;
-        opC.expectPrepare().expectCommit();
+        MockTxOp::shared_ptr opA(new MockTxOp());
+        opA->expectPrepare().expectCommit();
+        MockTxOp::shared_ptr opB(new MockTxOp());
+        opB->expectPrepare().expectPrepare().expectCommit().expectCommit();//opB enlisted twice to test reative order
+        MockTxOp::shared_ptr opC(new MockTxOp());
+        opC->expectPrepare().expectCommit();
 
         TxBuffer buffer;
-        buffer.enlist(&opA);
-        buffer.enlist(&opB);
-        buffer.enlist(&opB);//opB enlisted twice
-        buffer.enlist(&opC);
+        buffer.enlist(static_pointer_cast<TxOp>(opA));
+        buffer.enlist(static_pointer_cast<TxOp>(opB));
+        buffer.enlist(static_pointer_cast<TxOp>(opB));//opB enlisted twice
+        buffer.enlist(static_pointer_cast<TxOp>(opC));
 
         CPPUNIT_ASSERT(buffer.prepare(&store));
         buffer.commit();
         store.check();
         CPPUNIT_ASSERT(store.isCommitted());
-        opA.check();
-        opB.check();
-        opC.check();
+        opA->check();
+        opB->check();
+        opC->check();
     }
 
     void testFailOnPrepare(){
         MockTransactionalStore store;
         store.expectBegin().expectAbort();
 
-        MockTxOp opA;
-        opA.expectPrepare();
-        MockTxOp opB(true);
-        opB.expectPrepare();
-        MockTxOp opC;//will never get prepare as b will fail
+        MockTxOp::shared_ptr opA(new MockTxOp());
+        opA->expectPrepare();
+        MockTxOp::shared_ptr opB(new MockTxOp(true));
+        opB->expectPrepare();
+        MockTxOp::shared_ptr opC(new MockTxOp());//will never get prepare as b will fail
 
         TxBuffer buffer;
-        buffer.enlist(&opA);
-        buffer.enlist(&opB);
-        buffer.enlist(&opC);
+        buffer.enlist(static_pointer_cast<TxOp>(opA));
+        buffer.enlist(static_pointer_cast<TxOp>(opB));
+        buffer.enlist(static_pointer_cast<TxOp>(opC));
 
         CPPUNIT_ASSERT(!buffer.prepare(&store));
         store.check();
         CPPUNIT_ASSERT(store.isAborted());
-        opA.check();
-        opB.check();
-        opC.check();
+        opA->check();
+        opB->check();
+        opC->check();
     }
 
     void testRollback(){
-        MockTxOp opA;
-        opA.expectRollback();
-        MockTxOp opB(true);
-        opB.expectRollback();
-        MockTxOp opC;
-        opC.expectRollback();
+        MockTxOp::shared_ptr opA(new MockTxOp());
+        opA->expectRollback();
+        MockTxOp::shared_ptr opB(new MockTxOp(true));
+        opB->expectRollback();
+        MockTxOp::shared_ptr opC(new MockTxOp());
+        opC->expectRollback();
 
         TxBuffer buffer;
-        buffer.enlist(&opA);
-        buffer.enlist(&opB);
-        buffer.enlist(&opC);
+        buffer.enlist(static_pointer_cast<TxOp>(opA));
+        buffer.enlist(static_pointer_cast<TxOp>(opB));
+        buffer.enlist(static_pointer_cast<TxOp>(opC));
 
         buffer.rollback();
-        opA.check();
-        opB.check();
-        opC.check();
+        opA->check();
+        opB->check();
+        opC->check();
     }
 
     void testBufferIsClearedAfterRollback(){
-        MockTxOp opA;
-        opA.expectRollback();
-        MockTxOp opB;
-        opB.expectRollback();
+        MockTxOp::shared_ptr opA(new MockTxOp());
+        opA->expectRollback();
+        MockTxOp::shared_ptr opB(new MockTxOp());
+        opB->expectRollback();
 
         TxBuffer buffer;
-        buffer.enlist(&opA);
-        buffer.enlist(&opB);
+        buffer.enlist(static_pointer_cast<TxOp>(opA));
+        buffer.enlist(static_pointer_cast<TxOp>(opB));
 
         buffer.rollback();
         buffer.commit();//second call should not reach ops
-        opA.check();
-        opB.check();
+        opA->check();
+        opB->check();
     }
 
     void testBufferIsClearedAfterCommit(){
-        MockTxOp opA;
-        opA.expectCommit();
-        MockTxOp opB;
-        opB.expectCommit();
+        MockTxOp::shared_ptr opA(new MockTxOp());
+        opA->expectCommit();
+        MockTxOp::shared_ptr opB(new MockTxOp());
+        opB->expectCommit();
 
         TxBuffer buffer;
-        buffer.enlist(&opA);
-        buffer.enlist(&opB);
+        buffer.enlist(static_pointer_cast<TxOp>(opA));
+        buffer.enlist(static_pointer_cast<TxOp>(opB));
 
         buffer.commit();
         buffer.rollback();//second call should not reach ops
-        opA.check();
-        opB.check();
+        opA->check();
+        opB->check();
     }
 };