QPID-3603: Separate QueueGuard from ReplicatingSubscription.

QueueGuard: implements QueueObserver to delay completion of new messages.
ReplicatingSubscription: Implements subscription, sends messages & events to backup.

These were previously combined as one. QueueGuard is now separated out
so that it can be created before the ReplicatingSubscription, in
anticipation of an expected backup connecting. This is needed for 2 reasons:
- new queues must be guarded until they are backuped up.
- after a failover, all queues must be guarded until backups are ready.

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

1 files changed, 13 insertions, 11 deletions

diff --git a/qpid/cpp/src/qpid/ha/Primary.h b/qpid/cpp/src/qpid/ha/Primary.h
index 7e347fdbe2..3a1a9be9e8 100644
--- a/qpid/cpp/src/qpid/ha/Primary.h
+++ b/qpid/cpp/src/qpid/ha/Primary.h
@@ -22,7 +22,8 @@
  *
  */
 
-#include "LogPrefix.h"
+#include "UnreadyQueueSet.h"
+#include "types.h"
 #include "qpid/sys/Mutex.h"
 #include <boost/shared_ptr.hpp>
 #include <map>
@@ -36,10 +37,11 @@ class Queue;
 
 namespace ha {
 class HaBroker;
+class ReplicatingSubscription;
 
 /**
- * State associated with a primary broker.  Tracks replicating
- * subscriptions to determine when primary is ready.
+ * State associated with a primary broker. Tracks replicating
+ * subscriptions to determine when primary is active.
  *
  * THREAD SAFE: readyReplica is called in arbitray threads.
  */
@@ -47,22 +49,22 @@ class Primary
 {
   public:
     static Primary* get() { return instance; }
-    Primary(HaBroker& b);
 
-    void readyReplica(const std::string& q);
-    void removeReplica(const std::string& q);
+    Primary(HaBroker& hb, const IdSet& expectedBackups);
 
-  private:
-    typedef std::map<std::string, size_t> QueueCounts;
+    void readyReplica(const ReplicatingSubscription&);
+    void removeReplica(const std::string& q);
 
-    void activate(sys::Mutex::ScopedLock&);
+    UnreadyQueueSet& getUnreadyQueueSet() { return queues; }
+    bool isActive() { return activated; }
 
+  private:
     sys::Mutex lock;
     HaBroker& haBroker;
-    LogPrefix logPrefix;
-    QueueCounts queues;
+    std::string logPrefix;
     size_t expected, unready;
     bool activated;
+    UnreadyQueueSet queues;
 
     static Primary* instance;
 };