Enable dequeue for prototype cluster

 - qpid/broker/SemanticState.cpp: moved doOutput into write idle callback.
 - qpid/broker/Connection.cpp: make doOutput an intercept point.
 - qpid/cluster/*: intercept doOutput to serialize output in cluster thread.


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

1 files changed, 11 insertions, 10 deletions

diff --git a/cpp/src/qpid/cluster/ConnectionInterceptor.h b/cpp/src/qpid/cluster/ConnectionInterceptor.h
index d499acb832..7a955ddd80 100644
--- a/cpp/src/qpid/cluster/ConnectionInterceptor.h
+++ b/cpp/src/qpid/cluster/ConnectionInterceptor.h
@@ -38,17 +38,16 @@ class ConnectionInterceptor {
     ConnectionInterceptor(broker::Connection&, Cluster&,
                           Cluster::ShadowConnectionId shadowId=Cluster::ShadowConnectionId(0,0));
     ~ConnectionInterceptor();
+    
+    Cluster::ShadowConnectionId getShadowId() const { return shadowId; }
 
-    // Called on self-delivery
-    void deliver(framing::AMQFrame& f);
+    bool isLocal() const { return shadowId == Cluster::ShadowConnectionId(0,0); }
 
-    // Called on self-delivery of my own cluster.connection-close
+    // self-delivery of intercepted extension points.
+    void deliver(framing::AMQFrame& f);
     void deliverClosed();
+    void deliverDoOutput();
 
-    Cluster::ShadowConnectionId getShadowId() const { return shadowId; }
-
-    bool isLocal() const { return shadowId == Cluster::ShadowConnectionId(0,0); }
-    
   private:
     struct NullConnectionHandler : public qpid::sys::ConnectionOutputHandler {
         void close() {}
@@ -57,12 +56,14 @@ class ConnectionInterceptor {
         void activateOutput() {}
     };
     
-    // Functions to add to Connection extension points.
+    // Functions to intercept to Connection extension points.
     void received(framing::AMQFrame&);
     void closed();
+    bool doOutput();
 
-    boost::function<void(framing::AMQFrame&)> receivedNext;
-    boost::function<void()> closedNext;
+    boost::function<void (framing::AMQFrame&)> receivedNext;
+    boost::function<void ()> closedNext;
+    boost::function<bool ()> doOutputNext;
 
     boost::intrusive_ptr<broker::Connection> connection;
     Cluster& cluster;