1 files changed, 15 insertions, 0 deletions
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