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

7 files changed, 50 insertions, 25 deletions

diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index e77911bd10..2525fed864 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -51,6 +51,7 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
     ConnectionState(out_, broker_),
     receivedFn(boost::bind(&Connection::receivedImpl, this, _1)),
     closedFn(boost::bind(&Connection::closedImpl, this)),
+    doOutputFn(boost::bind(&Connection::doOutputImpl, this)),
     adapter(*this, isLink_),
     isLink(isLink_),
     mgmtClosing(false),
@@ -192,8 +193,9 @@ void Connection::closedImpl(){ // Physically closed, suspend open sessions.
     }
 }
 
-bool Connection::doOutput()
-{    
+bool Connection::doOutput() { return doOutputFn(); }
+
+bool Connection::doOutputImpl() {    
     try{
         if (ioCallback)
             ioCallback(); // Lend the IO thread for management processing
diff --git a/cpp/src/qpid/broker/Connection.h b/cpp/src/qpid/broker/Connection.h
index 0d646bab83..ae8708861a 100644
--- a/cpp/src/qpid/broker/Connection.h
+++ b/cpp/src/qpid/broker/Connection.h
@@ -96,7 +96,8 @@ class Connection : public sys::ConnectionInputHandler,
 
     // Extension points: allow plugins to insert additional functionality.
     boost::function<void(framing::AMQFrame&)> receivedFn;
-    boost::function<void()> closedFn; 
+    boost::function<void ()> closedFn;
+    boost::function<bool ()> doOutputFn;
 
   private:
     typedef boost::ptr_map<framing::ChannelId, SessionHandler> ChannelMap;
@@ -104,6 +105,7 @@ class Connection : public sys::ConnectionInputHandler,
 
     void receivedImpl(framing::AMQFrame& frame);
     void closedImpl();
+    bool doOutputImpl();
 
     ChannelMap channels;
     framing::AMQP_ClientProxy::Connection* client;
diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp
index 2bc366dc86..085578295d 100644
--- a/cpp/src/qpid/broker/SemanticState.cpp
+++ b/cpp/src/qpid/broker/SemanticState.cpp
@@ -380,9 +380,13 @@ void SemanticState::requestDispatch()
 
 void SemanticState::requestDispatch(ConsumerImpl& c)
 {    
-    if(c.isBlocked()) {
-        c.doOutput();
-    }
+    if(c.isBlocked())
+        outputTasks.activateOutput();
+    // TODO aconway 2008-07-16:  we could directly call
+    //  c.doOutput();
+    // since we are in the connections thread but for consistency
+    // activateOutput() will set it up to be called in the next write idle.
+    // Current cluster code depends on this, review cluster code to change.
 }
 
 void SemanticState::complete(DeliveryRecord& delivery)
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;
diff --git a/cpp/src/tests/cluster_test.cpp b/cpp/src/tests/cluster_test.cpp
index cafac489d2..567896d44d 100644
--- a/cpp/src/tests/cluster_test.cpp
+++ b/cpp/src/tests/cluster_test.cpp
@@ -225,10 +225,6 @@ QPID_AUTO_TEST_CASE(testMessageEnqueue) {
     BOOST_CHECK_EQUAL(string("bar"), msg.getData());
 }
 
-#if 0
-
-// FIXME aconway 2008-07-16: Implement cluster dequeue notification, enable this test.
-
 QPID_AUTO_TEST_CASE(testMessageDequeue) {
     // Enqueue on one broker, dequeue on two others.
     ClusterFixture cluster (3);
@@ -236,10 +232,10 @@ QPID_AUTO_TEST_CASE(testMessageDequeue) {
     c0.session.queueDeclare("q");
     c0.session.messageTransfer(arg::content=TransferContent("foo", "q"));
     c0.session.messageTransfer(arg::content=TransferContent("bar", "q"));
-    c0.session.close();
 
     Message msg;
 
+    // Dequeue on 2 others, ensure correct order.
     Client c1(cluster[1]);
     BOOST_CHECK(c1.subs.get(msg, "q"));
     BOOST_CHECK_EQUAL("foo", msg.getData());
@@ -247,12 +243,13 @@ QPID_AUTO_TEST_CASE(testMessageDequeue) {
     Client c2(cluster[2]);
     BOOST_CHECK(c1.subs.get(msg, "q"));
     BOOST_CHECK_EQUAL("bar", msg.getData());
-    QueueQueryResult r = c2.session.queueQuery("q");
-    BOOST_CHECK_EQUAL(0u, r.getMessageCount());
+
+    // Queue should be empty on all queues.
+    BOOST_CHECK_EQUAL(0u, c0.session.queueQuery("q").getMessageCount());
+    BOOST_CHECK_EQUAL(0u, c1.session.queueQuery("q").getMessageCount());
+    BOOST_CHECK_EQUAL(0u, c2.session.queueQuery("q").getMessageCount());
 }
 
 // TODO aconway 2008-06-25: failover.
 
-#endif
-
 QPID_AUTO_TEST_SUITE_END()