Refactored cluster to intercept at ConnectionCode, using sys:: interfaces rather than boost functions.

Use framing::Operations and Invoker to dispatch cluster methods.


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

6 files changed, 25 insertions, 34 deletions

diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index 4d7c07649b..e983aee5c9 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -133,7 +133,7 @@ Broker::Broker(const Broker::Options& conf) :
 	acl(0),
     dataDir(conf.noDataDir ? std::string () : conf.dataDir),
     links(this),
-    factory(*this),
+    factory(new ConnectionFactory(*this)),
     sessionManager(
         qpid::SessionState::Configuration(
             conf.replayFlushLimit*1024, // convert kb to bytes.
@@ -372,7 +372,7 @@ uint16_t Broker::getPort() const  {
 // TODO: This should iterate over all protocolFactories
 void Broker::accept() {
     for (unsigned int i = 0; i < protocolFactories.size(); ++i)
-        protocolFactories[i]->accept(poller, &factory);
+        protocolFactories[i]->accept(poller, factory.get());
 }
 
 
@@ -382,7 +382,7 @@ void Broker::connect(
     boost::function2<void, int, std::string> failed,
     sys::ConnectionCodec::Factory* f)
 {
-    getProtocolFactory()->connect(poller, host, port, f ? f : &factory, failed);
+    getProtocolFactory()->connect(poller, host, port, f ? f : factory.get(), failed);
 }
 
 void Broker::connect(
diff --git a/cpp/src/qpid/broker/Broker.h b/cpp/src/qpid/broker/Broker.h
index f7399c375f..bd17d1bb00 100644
--- a/cpp/src/qpid/broker/Broker.h
+++ b/cpp/src/qpid/broker/Broker.h
@@ -103,7 +103,7 @@ class Broker : public sys::Runnable, public Plugin::Target,
     QueueRegistry queues;
     ExchangeRegistry exchanges;
     LinkRegistry links;
-    ConnectionFactory factory;
+    boost::shared_ptr<sys::ConnectionCodec::Factory> factory;
     DtxManager dtxManager;
     SessionManager sessionManager;
     management::ManagementAgent* managementAgent;
@@ -178,7 +178,10 @@ class Broker : public sys::Runnable, public Plugin::Target,
     boost::shared_ptr<sys::ProtocolFactory> getProtocolFactory() const;
 
     /** Expose poller so plugins can register their descriptors. */
-    boost::shared_ptr<sys::Poller> getPoller(); 
+    boost::shared_ptr<sys::Poller> getPoller();
+
+    boost::shared_ptr<sys::ConnectionCodec::Factory> getConnectionFactory() { return factory; }
+    void setConnectionFactory(boost::shared_ptr<sys::ConnectionCodec::Factory> f) { factory = f; }
 };
 
 }}
diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index d65dbaeec7..8ed3ce84c8 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -49,9 +49,6 @@ namespace broker {
 
 Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std::string& mgmtId_, bool isLink_) :
     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),
@@ -72,8 +69,6 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
             mgmtObject = new management::Connection(agent, this, parent, mgmtId, !isLink);
         agent->addObject(mgmtObject);
     }
-
-    Plugin::initializeAll(*this); // Let plug-ins update extension points.
 }
 
 void Connection::requestIOProcessing(boost::function0<void> callback)
@@ -90,9 +85,7 @@ Connection::~Connection()
         links.notifyClosed(mgmtId);
 }
 
-void Connection::received(framing::AMQFrame& frame) { receivedFn(frame); }
-
-void Connection::receivedImpl(framing::AMQFrame& frame){
+void Connection::received(framing::AMQFrame& frame) {
     if (frame.getChannel() == 0 && frame.getMethod()) {
         adapter.handle(frame);
     } else {
@@ -172,9 +165,7 @@ void Connection::idleOut(){}
 
 void Connection::idleIn(){}
 
-void Connection::closed() { closedFn(); }
-
-void Connection::closedImpl(){ // Physically closed, suspend open sessions.
+void Connection::closed(){ // Physically closed, suspend open sessions.
     try {
         while (!channels.empty()) 
             ptr_map_ptr(channels.begin())->handleDetach();
@@ -194,9 +185,7 @@ void Connection::closedImpl(){ // Physically closed, suspend open sessions.
 
 bool Connection::hasOutput() { return outputTasks.hasOutput(); }
 
-bool Connection::doOutput() { return doOutputFn(); }
-
-bool Connection::doOutputImpl() {    
+bool Connection::doOutput() {    
     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 1367f3b9ca..8c23e64edf 100644
--- a/cpp/src/qpid/broker/Connection.h
+++ b/cpp/src/qpid/broker/Connection.h
@@ -44,7 +44,6 @@
 #include "SessionHandler.h"
 #include "qpid/management/Manageable.h"
 #include "qpid/management/Connection.h"
-#include "qpid/Plugin.h"
 #include "qpid/RefCounted.h"
 
 #include <boost/ptr_container/ptr_map.hpp>
@@ -56,7 +55,6 @@ class LinkRegistry;
 
 class Connection : public sys::ConnectionInputHandler, 
                    public ConnectionState,
-                   public Plugin::Target,
                    public RefCounted
 {
   public:
@@ -95,19 +93,10 @@ class Connection : public sys::ConnectionInputHandler,
     void notifyConnectionForced(const std::string& text);
     void setUserId(const string& uid);
 
-    // Extension points: allow plugins to insert additional functionality.
-    boost::function<void(framing::AMQFrame&)> receivedFn;
-    boost::function<void ()> closedFn;
-    boost::function<bool ()> doOutputFn;
-
   private:
     typedef boost::ptr_map<framing::ChannelId, SessionHandler> ChannelMap;
     typedef std::vector<Queue::shared_ptr>::iterator queue_iterator;
 
-    void receivedImpl(framing::AMQFrame& frame);
-    void closedImpl();
-    bool doOutputImpl();
-
     ChannelMap channels;
     framing::AMQP_ClientProxy::Connection* client;
     ConnectionHandler adapter;
diff --git a/cpp/src/qpid/broker/ConnectionFactory.cpp b/cpp/src/qpid/broker/ConnectionFactory.cpp
index 5de5a0230a..e6d8c49055 100644
--- a/cpp/src/qpid/broker/ConnectionFactory.cpp
+++ b/cpp/src/qpid/broker/ConnectionFactory.cpp
@@ -21,11 +21,14 @@
 #include "ConnectionFactory.h"
 #include "qpid/framing/ProtocolVersion.h"
 #include "qpid/amqp_0_10/Connection.h"
+#include "qpid/broker/Connection.h"
 
 namespace qpid {
 namespace broker {
 
 using framing::ProtocolVersion;
+typedef std::auto_ptr<amqp_0_10::Connection> ConnectionPtr;
+typedef std::auto_ptr<sys::ConnectionInputHandler> InputPtr;
 
 ConnectionFactory::ConnectionFactory(Broker& b) : broker(b) {}
 
@@ -33,15 +36,21 @@ ConnectionFactory::~ConnectionFactory() {}
 
 sys::ConnectionCodec*
 ConnectionFactory::create(ProtocolVersion v, sys::OutputControl& out, const std::string& id) {
-    if (v == ProtocolVersion(0, 10))
-        return new amqp_0_10::Connection(out, broker, id);
+    if (v == ProtocolVersion(0, 10)) {
+        ConnectionPtr c(new amqp_0_10::Connection(out, id, false));
+        c->setInputHandler(InputPtr(new broker::Connection(c.get(), broker, id, false)));
+        return c.release();
+    }
     return 0;
 }
 
 sys::ConnectionCodec*
 ConnectionFactory::create(sys::OutputControl& out, const std::string& id) {
     // used to create connections from one broker to another
-    return new amqp_0_10::Connection(out, broker, id, true);
+    ConnectionPtr c(new amqp_0_10::Connection(out, id, true));
+    c->setInputHandler(InputPtr(new broker::Connection(c.get(), broker, id, true)));
+    return c.release();
 }
 
+    
 }} // namespace qpid::broker
diff --git a/cpp/src/qpid/broker/ConnectionFactory.h b/cpp/src/qpid/broker/ConnectionFactory.h
index 5797495054..c61da81024 100644
--- a/cpp/src/qpid/broker/ConnectionFactory.h
+++ b/cpp/src/qpid/broker/ConnectionFactory.h
@@ -27,7 +27,8 @@ namespace qpid {
 namespace broker {
 class Broker;
 
-class ConnectionFactory : public sys::ConnectionCodec::Factory {
+class ConnectionFactory : public sys::ConnectionCodec::Factory
+{
   public:
     ConnectionFactory(Broker& b);