Pre-buffering output strategy for cluster.

Additional hooks in broker code, should not affect standalone broker.


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

2 files changed, 11 insertions, 10 deletions

diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index ab18d1f035..d65dbaeec7 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -79,7 +79,7 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
 void Connection::requestIOProcessing(boost::function0<void> callback)
 {
     ioCallback = callback;
-    out->activateOutput();
+    out.activateOutput();
 }
 
 Connection::~Connection()
@@ -178,7 +178,6 @@ void Connection::closedImpl(){ // Physically closed, suspend open sessions.
     try {
         while (!channels.empty()) 
             ptr_map_ptr(channels.begin())->handleDetach();
-        // FIXME aconway 2008-07-15: exclusive is per-session not per-connection in 0-10.
         while (!exclusiveQueues.empty()) {
             Queue::shared_ptr q(exclusiveQueues.front());
             q->releaseExclusiveOwnership();
@@ -245,7 +244,7 @@ Manageable::status_t Connection::ManagementMethod(uint32_t methodId, Args&)
     case management::Connection::METHOD_CLOSE :
         mgmtClosing = true;
         if (mgmtObject != 0) mgmtObject->set_closing(1);
-        out->activateOutput();
+        out.activateOutput();
         status = Manageable::STATUS_OK;
         break;
     }
diff --git a/cpp/src/qpid/broker/ConnectionState.h b/cpp/src/qpid/broker/ConnectionState.h
index c9cf6ece8d..97055f8b2e 100644
--- a/cpp/src/qpid/broker/ConnectionState.h
+++ b/cpp/src/qpid/broker/ConnectionState.h
@@ -24,7 +24,7 @@
 #include <vector>
 
 #include "qpid/sys/AggregateOutput.h"
-#include "qpid/sys/ConnectionOutputHandler.h"
+#include "qpid/sys/ConnectionOutputHandlerPtr.h"
 #include "qpid/framing/ProtocolVersion.h"
 #include "qpid/management/Manageable.h"
 #include "Broker.h"
@@ -34,11 +34,14 @@ namespace broker {
 
 class ConnectionState : public ConnectionToken, public management::Manageable
 {
+  protected:
+    sys::ConnectionOutputHandlerPtr out;
+
   public:
-    ConnectionState(qpid::sys::ConnectionOutputHandler* o, Broker& b) : 
+    ConnectionState(qpid::sys::ConnectionOutputHandler* o, Broker& b) :
+        out(o),
         broker(b), 
-        outputTasks(*o),
-        out(o), 
+        outputTasks(out),
         framemax(65535), 
         heartbeat(0),
         stagingThreshold(broker.getStagingThreshold())
@@ -67,14 +70,13 @@ class ConnectionState : public ConnectionToken, public management::Manageable
     //contained output tasks
     sys::AggregateOutput outputTasks;
 
-    sys::ConnectionOutputHandler& getOutput() const { return *out; }
+    sys::ConnectionOutputHandlerPtr& getOutput() { return out; }
     framing::ProtocolVersion getVersion() const { return version; }
 
-    void setOutputHandler(qpid::sys::ConnectionOutputHandler* o) { out = o; }
+    void setOutputHandler(qpid::sys::ConnectionOutputHandler* o) { out.set(o); }
 
   protected:
     framing::ProtocolVersion version;
-    sys::ConnectionOutputHandler* out;
     uint32_t framemax;
     uint16_t heartbeat;
     uint64_t stagingThreshold;