HandlerChain: plug-in handler chain extension points. Replaces Handler<T>::Chain.

Updated Sessoin & Connection handler chains and Cluster.


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

9 files changed, 15 insertions, 151 deletions

diff --git a/cpp/src/qpid/broker/Broker.h b/cpp/src/qpid/broker/Broker.h
index 9a9f502bf0..be59cef24c 100644
--- a/cpp/src/qpid/broker/Broker.h
+++ b/cpp/src/qpid/broker/Broker.h
@@ -23,7 +23,6 @@
  */
 
 #include "ConnectionFactory.h"
-#include "ConnectionManager.h"
 #include "ConnectionToken.h"
 #include "DirectExchange.h"
 #include "DtxManager.h"
@@ -121,7 +120,6 @@ class Broker : public sys::Runnable, public Plugin::Target,
     Options& getOptions() { return config; }
 
     SessionManager& getSessionManager() { return sessionManager; }
-    ConnectionManager& getConnectionManager() { return connectionManager; }
 
     management::ManagementObject*     GetManagementObject (void) const;
     management::Manageable*           GetVhostObject      (void) const;
@@ -159,7 +157,6 @@ class Broker : public sys::Runnable, public Plugin::Target,
     ConnectionFactory factory;
     DtxManager dtxManager;
     SessionManager sessionManager;
-    ConnectionManager connectionManager;
     management::ManagementAgent* managementAgent;
     management::Broker*          mgmtObject;
     Vhost::shared_ptr            vhostObject;
diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index bb99c61cdd..61384638b3 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -88,7 +88,7 @@ Connection::~Connection()
         links.notifyClosed(mgmtId);
 }
 
-void Connection::received(framing::AMQFrame& frame){ inChain(frame); }
+void Connection::received(framing::AMQFrame& frame){ inChain->handle(frame); }
     
 void Connection::receivedLast(framing::AMQFrame& frame){
     if (frame.getChannel() == 0 && frame.getMethod()) {
diff --git a/cpp/src/qpid/broker/Connection.h b/cpp/src/qpid/broker/Connection.h
index 717e1a6270..c911e88200 100644
--- a/cpp/src/qpid/broker/Connection.h
+++ b/cpp/src/qpid/broker/Connection.h
@@ -43,6 +43,7 @@
 #include "SessionHandler.h"
 #include "qpid/management/Manageable.h"
 #include "qpid/management/Connection.h"
+#include "qpid/HandlerChain.h"
 
 #include <boost/ptr_container/ptr_map.hpp>
 
@@ -91,8 +92,6 @@ class Connection : public sys::ConnectionInputHandler,
     void notifyConnectionForced(const std::string& text);
     void setUserId(const string& uid);
 
-    framing::FrameHandler::Chain& getInChain() { return inChain; } 
-
   private:
     typedef boost::ptr_map<framing::ChannelId, SessionHandler> ChannelMap;
     typedef std::vector<Queue::shared_ptr>::iterator queue_iterator;
@@ -110,8 +109,7 @@ class Connection : public sys::ConnectionInputHandler,
     management::Connection* mgmtObject;
     LinkRegistry& links;
     framing::FrameHandler::MemFunRef<Connection, &Connection::receivedLast> lastInHandler;
-    framing::FrameHandler::Chain inChain;
-
+    PluginHandlerChain<framing::FrameHandler, Connection> inChain;
 };
 
 }}
diff --git a/cpp/src/qpid/broker/ConnectionManager.cpp b/cpp/src/qpid/broker/ConnectionManager.cpp
deleted file mode 100644
index 165de5220e..0000000000
--- a/cpp/src/qpid/broker/ConnectionManager.cpp
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-#include "ConnectionManager.h"
-#include "Connection.h"
-
-namespace qpid {
-namespace broker {
-
-std::auto_ptr<Connection>
-ConnectionManager::create(sys::ConnectionOutputHandler* out, Broker& broker, const std::string& mgmtId, bool isClient) {
-    std::auto_ptr<Connection> c(new Connection(out, broker, mgmtId, isClient));
-    sys::Mutex::ScopedLock l(lock);
-    std::for_each(observers.begin(), observers.end(),
-                  boost::bind(&Observer::created, _1, boost::ref(*c)));
-    return c;
-}
-
-void ConnectionManager::add(const boost::intrusive_ptr<Observer>& observer) {
-    sys::Mutex::ScopedLock l(lock);
-    observers.push_back(observer);
-}
-
-}} // namespace qpid::broker
diff --git a/cpp/src/qpid/broker/ConnectionManager.h b/cpp/src/qpid/broker/ConnectionManager.h
deleted file mode 100644
index a999523d0d..0000000000
--- a/cpp/src/qpid/broker/ConnectionManager.h
+++ /dev/null
@@ -1,70 +0,0 @@
-#ifndef QPID_BROKER_CONNECTIONMANAGER_H
-#define QPID_BROKER_CONNECTIONMANAGER_H
-
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "qpid/RefCounted.h"
-#include "qpid/sys/Mutex.h"
-#include <boost/intrusive_ptr.hpp>
-#include <vector>
-#include <memory>
-
-namespace qpid {
-
-namespace sys {
-class ConnectionOutputHandler;
-}
-
-namespace broker {
-
-class Broker;
-class Connection;
-
-/**
- * Manages connections and observers.
- */
-class ConnectionManager {
-  public:
-
-    /**
-     * Observer notified of ConnectionManager events.
-     */
-    struct Observer : public RefCounted {
-        /** Called when a connection is attached. */
-        virtual void created(Connection&) {}
-    };
-
-    /** Called to create a new Connection, applies observers. */
-    std::auto_ptr<Connection> create(sys::ConnectionOutputHandler* out, Broker& broker, const std::string& mgmtId, bool isClient = false);
-
-    /** Add an observer */
-    void add(const boost::intrusive_ptr<Observer>&);
-
-  private:
-    typedef std::vector<boost::intrusive_ptr<Observer> > Observers;
-
-    sys::Mutex lock;
-    Observers observers;
-};
-}} // namespace qpid::broker
-
-#endif  /*!QPID_BROKER_CONNECTIONMANAGER_H*/
diff --git a/cpp/src/qpid/broker/SessionManager.cpp b/cpp/src/qpid/broker/SessionManager.cpp
index 69ef29c3eb..e7190fdae6 100644
--- a/cpp/src/qpid/broker/SessionManager.cpp
+++ b/cpp/src/qpid/broker/SessionManager.cpp
@@ -55,11 +55,8 @@ std::auto_ptr<SessionState>  SessionManager::attach(SessionHandler& h, const Ses
         throw SessionBusyException(QPID_MSG("Session already attached: " << id));
     Detached::iterator i = std::find(detached.begin(), detached.end(), id);
     std::auto_ptr<SessionState> state;
-    if (i == detached.end())  {
+    if (i == detached.end())  
         state.reset(new SessionState(broker, h, id, config));
-    for_each(observers.begin(), observers.end(),
-                 boost::bind(&Observer::opened, _1,boost::ref(*state)));
-    }
     else {
         state.reset(detached.release(i).release());
         state->attach(h);
@@ -99,8 +96,4 @@ void SessionManager::eraseExpired() {
     }
 }
 
-void SessionManager::add(const intrusive_ptr<Observer>& o) {
-    observers.push_back(o);
-}
-
 }} // namespace qpid::broker
diff --git a/cpp/src/qpid/broker/SessionManager.h b/cpp/src/qpid/broker/SessionManager.h
index 9a4142f613..db88e7ec10 100644
--- a/cpp/src/qpid/broker/SessionManager.h
+++ b/cpp/src/qpid/broker/SessionManager.h
@@ -46,14 +46,6 @@ class SessionHandler;
  */
 class SessionManager : private boost::noncopyable {
   public:
-    /**
-     * Observer notified of SessionManager events.
-     */
-    struct Observer : public RefCounted {
-        /** Called when a stateless session is attached. */
-        virtual void opened(SessionState&) {}
-    };
-    
     SessionManager(const qpid::SessionState::Configuration&, Broker&);
     
     ~SessionManager();
@@ -67,9 +59,6 @@ class SessionManager : private boost::noncopyable {
     /** Forget about an attached session. Called by SessionState destructor. */
     void forget(const SessionId&);
 
-    /** Add an Observer. */
-    void add(const boost::intrusive_ptr<Observer>&);
-    
     Broker& getBroker() const { return broker; }
 
     const qpid::SessionState::Configuration& getSessionConfig() const { return config; }
@@ -77,7 +66,6 @@ class SessionManager : private boost::noncopyable {
   private:
     typedef boost::ptr_vector<SessionState> Detached; // Sorted in expiry order.
     typedef std::set<SessionId> Attached;
-    typedef std::vector<boost::intrusive_ptr<Observer> > Observers;
 
     void eraseExpired();             
 
@@ -85,7 +73,6 @@ class SessionManager : private boost::noncopyable {
     Detached detached;
     Attached attached;
     qpid::SessionState::Configuration config;
-    Observers observers;
     Broker& broker;
 };
 
diff --git a/cpp/src/qpid/broker/SessionState.cpp b/cpp/src/qpid/broker/SessionState.cpp
index 8a17a787a2..0a122fcae8 100644
--- a/cpp/src/qpid/broker/SessionState.cpp
+++ b/cpp/src/qpid/broker/SessionState.cpp
@@ -224,8 +224,8 @@ void SessionState::enqueued(boost::intrusive_ptr<Message> msg)
         getProxy().getMessage().accept(SequenceSet(msg->getCommandId()));        
 }
 
-void SessionState::handleIn(AMQFrame& f) { inChain.handle(f); }
-void SessionState::handleOut(AMQFrame& f) { outChain.handle(f); }
+void SessionState::handleIn(AMQFrame& f) { inChain->handle(f); }
+void SessionState::handleOut(AMQFrame& f) { outChain->handle(f); }
 
 void SessionState::handleInLast(AMQFrame& frame) {
     SequenceNumber commandId = receiverGetCurrent();
@@ -291,8 +291,4 @@ void SessionState::readyToSend() {
 
 Broker& SessionState::getBroker() { return broker; }
 
-framing::FrameHandler::Chain& SessionState::getInChain() { return inChain; }
-
-framing::FrameHandler::Chain& SessionState::getOutChain() { return outChain; }
-
 }} // namespace qpid::broker
diff --git a/cpp/src/qpid/broker/SessionState.h b/cpp/src/qpid/broker/SessionState.h
index 5d18ed161e..f6bf98d431 100644
--- a/cpp/src/qpid/broker/SessionState.h
+++ b/cpp/src/qpid/broker/SessionState.h
@@ -23,6 +23,7 @@
  */
 
 #include "qpid/SessionState.h"
+#include "qpid/HandlerChain.h"
 #include "qpid/framing/FrameHandler.h"
 #include "qpid/framing/SequenceSet.h"
 #include "qpid/sys/Mutex.h"
@@ -58,8 +59,8 @@ class SessionHandler;
 class SessionManager;
 
 /**
- * Broker-side session state includes sessions handler chains, which may
- * themselves have state. 
+ * Broker-side session state includes session's handler chains, which
+ * may themselves have state.
  */
 class SessionState : public qpid::SessionState, 
                      public SessionContext,
@@ -101,8 +102,9 @@ class SessionState : public qpid::SessionState,
 
     void readyToSend();
 
-    framing::FrameHandler::Chain& getInChain(); 
-    framing::FrameHandler::Chain& getOutChain(); 
+    // Tag types to identify PluginHandlerChains. 
+    struct InTag {};
+    struct OutTag {};
 
   private:
 
@@ -131,7 +133,9 @@ class SessionState : public qpid::SessionState,
     management::Session* mgmtObject;
     framing::FrameHandler::MemFunRef<SessionState, &SessionState::handleInLast> inLastHandler;
     framing::FrameHandler::MemFunRef<SessionState, &SessionState::handleOutLast> outLastHandler;
-    framing::FrameHandler::Chain inChain, outChain;
+
+    qpid::PluginHandlerChain<framing::FrameHandler, InTag> inChain;
+    qpid::PluginHandlerChain<framing::FrameHandler, OutTag> outChain;
 
   friend class SessionManager;
 };