QPID-4710: Add generic support for transactional dequeue to Queue class itself

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

5 files changed, 145 insertions, 0 deletions

diff --git a/qpid/cpp/src/CMakeLists.txt b/qpid/cpp/src/CMakeLists.txt
index e06468dddc..dbe58553a3 100644
--- a/qpid/cpp/src/CMakeLists.txt
+++ b/qpid/cpp/src/CMakeLists.txt
@@ -1124,6 +1124,8 @@ set (qpidbroker_SOURCES
      qpid/broker/TopicExchange.cpp
      qpid/broker/TxAccept.cpp
      qpid/broker/TxBuffer.cpp
+     qpid/broker/TxDequeue.h
+     qpid/broker/TxDequeue.cpp
      qpid/broker/Vhost.cpp
      qpid/broker/amqp_0_10/MessageTransfer.cpp
      qpid/management/ManagementAgent.cpp
diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index e12d689ba4..d897029c85 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/cpp/src/qpid/broker/Queue.cpp
@@ -36,6 +36,7 @@
 #include "qpid/broker/QueueRegistry.h"
 #include "qpid/broker/Selector.h"
 #include "qpid/broker/TransactionObserver.h"
+#include "qpid/broker/TxDequeue.h"
 
 //TODO: get rid of this
 #include "qpid/broker/amqp_0_10/MessageTransfer.h"
@@ -960,6 +961,24 @@ void Queue::dequeueFromStore(boost::intrusive_ptr<PersistableMessage> msg)
     }
 }
 
+void Queue::dequeue(const QueueCursor& cursor, TxBuffer* txn)
+{
+    if (txn) {
+        TxOp::shared_ptr op;
+        {
+            Mutex::ScopedLock locker(messageLock);
+            Message* msg = messages->find(cursor);
+            if (msg) {
+                op = TxOp::shared_ptr(new TxDequeue(cursor, shared_from_this(), msg->getSequence(), msg->getReplicationId()));
+            }
+        }
+        if (op) txn->enlist(op);
+    } else {
+        dequeue(0, cursor);
+    }
+}
+
+
 void Queue::dequeue(TransactionContext* ctxt, const QueueCursor& cursor)
 {
     ScopedUse u(barrier);
diff --git a/qpid/cpp/src/qpid/broker/Queue.h b/qpid/cpp/src/qpid/broker/Queue.h
index 407f262eac..65a91b8729 100644
--- a/qpid/cpp/src/qpid/broker/Queue.h
+++ b/qpid/cpp/src/qpid/broker/Queue.h
@@ -394,9 +394,16 @@ class Queue : public boost::enable_shared_from_this<Queue>,
     const QueueBindings& getBindings() const { return bindings; }
 
     /**
+     * Dequeue message referenced by cursor. If txn is specified, this will
+     * occur only when txn is committed.
+     */
+    QPID_BROKER_EXTERN void dequeue(const QueueCursor& cursor, TxBuffer* txn);
+
+    /**
      * dequeue from store (only done once messages is acknowledged)
      */
     QPID_BROKER_EXTERN void dequeue(TransactionContext* ctxt, const QueueCursor&);
+
     /**
      * Inform the queue that a previous transactional dequeue
      * committed.
diff --git a/qpid/cpp/src/qpid/broker/TxDequeue.cpp b/qpid/cpp/src/qpid/broker/TxDequeue.cpp
new file mode 100644
index 0000000000..bd83b434a9
--- /dev/null
+++ b/qpid/cpp/src/qpid/broker/TxDequeue.cpp
@@ -0,0 +1,65 @@
+/*
+ *
+ * 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 "qpid/broker/TxDequeue.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/broker/TransactionObserver.h"
+#include "qpid/log/Statement.h"
+
+namespace qpid {
+namespace broker {
+
+TxDequeue::TxDequeue(QueueCursor m, boost::shared_ptr<Queue> q,
+                     qpid::framing::SequenceNumber mId, qpid::framing::SequenceNumber rId)
+    : message(m), queue(q), messageId(mId), replicationId(rId) {}
+
+bool TxDequeue::prepare(TransactionContext* ctxt) throw()
+{
+    try{
+        queue->dequeue(ctxt, message);
+        return true;
+    }catch(const std::exception& e){
+        QPID_LOG(error, "Failed to prepare: " << e.what());
+        return false;
+    }catch(...){
+        QPID_LOG(error, "Failed to prepare");
+        return false;
+    }
+}
+
+void TxDequeue::commit() throw()
+{
+    try {
+        queue->dequeueCommitted(message);
+    } catch (const std::exception& e) {
+        QPID_LOG(error, "Failed to commit: " << e.what());
+    } catch(...) {
+        QPID_LOG(error, "Failed to commit (unknown error)");
+    }
+}
+
+void TxDequeue::rollback() throw() {}
+
+void TxDequeue::callObserver(const boost::shared_ptr<TransactionObserver>& observer)
+{
+    observer->dequeue(queue, messageId, replicationId);
+}
+
+}} // namespace qpid::broker
diff --git a/qpid/cpp/src/qpid/broker/TxDequeue.h b/qpid/cpp/src/qpid/broker/TxDequeue.h
new file mode 100644
index 0000000000..e312df9e31
--- /dev/null
+++ b/qpid/cpp/src/qpid/broker/TxDequeue.h
@@ -0,0 +1,52 @@
+#ifndef QPID_BROKER_TXDEQUEUE_H
+#define QPID_BROKER_TXDEQUEUE_H
+
+/*
+ *
+ * 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 "qpid/broker/QueueCursor.h"
+#include "qpid/broker/TxOp.h"
+#include "qpid/framing/SequenceNumber.h"
+
+namespace qpid {
+namespace broker {
+class Queue;
+
+/**
+ * Transaction dequeue operation
+ */
+class TxDequeue: public TxOp
+{
+  public:
+    TxDequeue(QueueCursor message, boost::shared_ptr<Queue> queue,
+              qpid::framing::SequenceNumber messageId, qpid::framing::SequenceNumber replicationId);
+    bool prepare(TransactionContext* ctxt) throw();
+    void commit() throw();
+    void rollback() throw();
+    void callObserver(const boost::shared_ptr<TransactionObserver>&);
+  private:
+    QueueCursor message;
+    boost::shared_ptr<Queue> queue;
+    qpid::framing::SequenceNumber messageId;
+    qpid::framing::SequenceNumber replicationId;
+};
+}} // namespace qpid::broker
+
+#endif  /*!QPID_BROKER_TXDEQUEUE_H*/