Merged out from trunkQPID-2519

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-2519@1187375 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 15 insertions, 14 deletions

diff --git a/cpp/src/qpid/broker/PersistableMessage.h b/cpp/src/qpid/broker/PersistableMessage.h
index 96fb922c1a..d29c2c45b4 100644
--- a/cpp/src/qpid/broker/PersistableMessage.h
+++ b/cpp/src/qpid/broker/PersistableMessage.h
@@ -31,6 +31,7 @@
 #include "qpid/framing/amqp_types.h"
 #include "qpid/sys/Mutex.h"
 #include "qpid/broker/PersistableQueue.h"
+#include "qpid/broker/AsyncCompletion.h"
 
 namespace qpid {
 namespace broker {
@@ -43,18 +44,19 @@ class MessageStore;
 class PersistableMessage : public Persistable
 {
     typedef std::list< boost::weak_ptr<PersistableQueue> > syncList;
-    sys::Mutex asyncEnqueueLock;
     sys::Mutex asyncDequeueLock;
     sys::Mutex storeLock;
-       
+
     /**
-     * Tracks the number of outstanding asynchronous enqueue
-     * operations. When the message is enqueued asynchronously the
-     * count is incremented; when that enqueue completes it is
-     * decremented. Thus when it is 0, there are no outstanding
-     * enqueues.
+     * "Ingress" messages == messages sent _to_ the broker.
+     * Tracks the number of outstanding asynchronous operations that must
+     * complete before an inbound message can be considered fully received by the
+     * broker.  E.g. all enqueues have completed, the message has been written
+     * to store, credit has been replenished, etc. Once all outstanding
+     * operations have completed, the transfer of this message from the client
+     * may be considered complete.
      */
-    int asyncEnqueueCounter;
+    AsyncCompletion ingressCompletion;
 
     /**
      * Tracks the number of outstanding asynchronous dequeue
@@ -65,7 +67,6 @@ class PersistableMessage : public Persistable
      */
     int asyncDequeueCounter;
 
-    void enqueueAsync();
     void dequeueAsync();
 
     syncList synclist;
@@ -80,8 +81,6 @@ class PersistableMessage : public Persistable
     ContentReleaseState contentReleaseState;
 
   protected:
-    /** Called when all enqueues are complete for this message. */
-    virtual void allEnqueuesComplete() = 0;
     /** Called when all dequeues are complete for this message. */
     virtual void allDequeuesComplete() = 0;
 
@@ -115,9 +114,12 @@ class PersistableMessage : public Persistable
 
     virtual QPID_BROKER_EXTERN bool isPersistent() const = 0;
 
-    QPID_BROKER_EXTERN bool isEnqueueComplete();
+    /** track the progress of a message received by the broker - see ingressCompletion above */
+    QPID_BROKER_INLINE_EXTERN bool isIngressComplete() { return ingressCompletion.isDone(); }
+    QPID_BROKER_INLINE_EXTERN AsyncCompletion& getIngressCompletion() { return ingressCompletion; }
 
-    QPID_BROKER_EXTERN void enqueueComplete();
+    QPID_BROKER_INLINE_EXTERN void enqueueStart() { ingressCompletion.startCompleter(); }
+    QPID_BROKER_INLINE_EXTERN void enqueueComplete() { ingressCompletion.finishCompleter(); }
 
     QPID_BROKER_EXTERN void enqueueAsync(PersistableQueue::shared_ptr queue,
                                          MessageStore* _store);
@@ -133,7 +135,6 @@ class PersistableMessage : public Persistable
     bool isStoredOnQueue(PersistableQueue::shared_ptr queue);
     
     void addToSyncList(PersistableQueue::shared_ptr queue, MessageStore* _store);
-    
 };
 
 }}