From 37103322e9fd5889bc2d412b64bb21eb8743391b Mon Sep 17 00:00:00 2001
From: Alan Conway <aconway@apache.org>
Date: Thu, 17 Jul 2008 02:06:50 +0000
Subject: 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
---
 cpp/src/qpid/cluster/Cluster.cpp               |  4 ++++
 cpp/src/qpid/cluster/ConnectionInterceptor.cpp | 15 +++++++++++++++
 cpp/src/qpid/cluster/ConnectionInterceptor.h   | 21 +++++++++++----------
 3 files changed, 30 insertions(+), 10 deletions(-)

(limited to 'cpp/src/qpid/cluster')

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 3b7f32e822..6623d1cde0 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -208,6 +208,10 @@ void Cluster::handleMethod(Id from, ConnectionInterceptor* connection, AMQMethod
           connection->deliverClosed();
           break;
       }
+      case CLUSTER_CONNECTION_DO_OUTPUT_METHOD_ID: {
+          connection->deliverDoOutput();
+          break;
+      }
       default:
         assert(0);
     }
diff --git a/cpp/src/qpid/cluster/ConnectionInterceptor.cpp b/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
index 5283ba9b1a..32c2054631 100644
--- a/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
+++ b/cpp/src/qpid/cluster/ConnectionInterceptor.cpp
@@ -20,6 +20,7 @@
  */
 #include "ConnectionInterceptor.h"
 #include "qpid/framing/ClusterConnectionCloseBody.h"
+#include "qpid/framing/ClusterConnectionDoOutputBody.h"
 #include "qpid/framing/AMQFrame.h"
 
 namespace qpid {
@@ -37,6 +38,7 @@ ConnectionInterceptor::ConnectionInterceptor(
     // Attach  my functions to Connection extension points.
     shift(receivedNext, connection->receivedFn, boost::bind(&ConnectionInterceptor::received, this, _1));
     shift(closedNext, connection->closedFn, boost::bind(&ConnectionInterceptor::closed, this));
+    shift(doOutputNext, connection->doOutputFn, boost::bind(&ConnectionInterceptor::doOutput, this));
 }
 
 ConnectionInterceptor::~ConnectionInterceptor() {
@@ -79,4 +81,17 @@ void ConnectionInterceptor::deliverClosed() {
     connection = 0;             
 }
 
+bool  ConnectionInterceptor::doOutput() {
+    cluster.send(AMQFrame(in_place<ClusterConnectionDoOutputBody>()), this);
+    return false;
+}
+
+void ConnectionInterceptor::deliverDoOutput() {
+    // FIXME aconway 2008-07-16: review thread safety.
+    // All connection processing happens in cluster queue, only read & write
+    // (from mutex-locked frameQueue) happens in reader/writer threads.
+    // 
+    doOutputNext();
+}
+
 }} // namespace qpid::cluster
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;
-- 
cgit v1.2.1