diff options
author | Alan Conway <aconway@apache.org> | 2012-02-17 14:12:47 +0000 |
---|---|---|
committer | Alan Conway <aconway@apache.org> | 2012-02-17 14:12:47 +0000 |
commit | bc4ca05a6928812d538c2fd60be27bbc91e7c283 (patch) | |
tree | 5291b1b160b204f546599f378e3c439287acf745 | |
parent | 2cba8c8e3030883e7b1079b813e521c6ac79e6a2 (diff) | |
download | qpid-python-bc4ca05a6928812d538c2fd60be27bbc91e7c283.tar.gz |
QPID-3603: Refactor LinkRegistry to use a ConnectionObserver.
git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid-3603-7@1245525 13f79535-47bb-0310-9956-ffa450edef68
-rw-r--r-- | qpid/cpp/src/qpid/broker/Broker.h | 2 | ||||
-rw-r--r-- | qpid/cpp/src/qpid/broker/Connection.cpp | 12 | ||||
-rw-r--r-- | qpid/cpp/src/qpid/broker/ConnectionObserver.h | 15 | ||||
-rw-r--r-- | qpid/cpp/src/qpid/broker/ConnectionObservers.h | 22 | ||||
-rw-r--r-- | qpid/cpp/src/qpid/broker/LinkRegistry.cpp | 12 | ||||
-rw-r--r-- | qpid/cpp/src/qpid/ha/ConnectionExcluder.h | 4 |
6 files changed, 45 insertions, 22 deletions
diff --git a/qpid/cpp/src/qpid/broker/Broker.h b/qpid/cpp/src/qpid/broker/Broker.h index 5d4cc99e66..fbe653bf5d 100644 --- a/qpid/cpp/src/qpid/broker/Broker.h +++ b/qpid/cpp/src/qpid/broker/Broker.h @@ -179,6 +179,7 @@ public: std::auto_ptr<MessageStore> store; AclModule* acl; DataDir dataDir; + ConnectionObservers connectionObservers; QueueRegistry queues; ExchangeRegistry exchanges; @@ -201,7 +202,6 @@ public: boost::intrusive_ptr<ExpiryPolicy> expiryPolicy; ConnectionCounter connectionCounter; ConsumerFactories consumerFactories; - ConnectionObservers connectionObservers; public: virtual ~Broker(); diff --git a/qpid/cpp/src/qpid/broker/Connection.cpp b/qpid/cpp/src/qpid/broker/Connection.cpp index 97e8e8ca13..1e6aab217c 100644 --- a/qpid/cpp/src/qpid/broker/Connection.cpp +++ b/qpid/cpp/src/qpid/broker/Connection.cpp @@ -104,8 +104,7 @@ Connection::Connection(ConnectionOutputHandler* out_, outboundTracker(*this) { outboundTracker.wrap(out); - if (link) - links.notifyConnection(mgmtId, this); + broker.getConnectionObservers().connection(*this); // In a cluster, allow adding the management object to be delayed. if (!delayManagement) addManagementObject(); if (!isShadow()) broker.getConnectionCounter().inc_connectionCount(); @@ -143,8 +142,7 @@ Connection::~Connection() if (!link && isClusterSafe()) agent->raiseEvent(_qmf::EventClientDisconnect(mgmtId, ConnectionState::getUserId())); } - if (link) - links.notifyClosed(mgmtId); + broker.getConnectionObservers().closed(*this); if (heartbeatTimer) heartbeatTimer->cancel(); @@ -165,8 +163,7 @@ void Connection::received(framing::AMQFrame& frame) { recordFromClient(frame); if (!wasOpen && isOpen()) { doIoCallbacks(); // Do any callbacks registered before we opened. - // FIXME aconway 2012-01-18: generic observer points. - broker.getConnectionObservers().connect(*this); + broker.getConnectionObservers().opened(*this); } } @@ -267,8 +264,7 @@ string Connection::getAuthCredentials() void Connection::notifyConnectionForced(const string& text) { - if (link) - links.notifyConnectionForced(mgmtId, text); + broker.getConnectionObservers().forced(*this, text); } void Connection::setUserId(const string& userId) diff --git a/qpid/cpp/src/qpid/broker/ConnectionObserver.h b/qpid/cpp/src/qpid/broker/ConnectionObserver.h index e59ec261bc..12aa8549fd 100644 --- a/qpid/cpp/src/qpid/broker/ConnectionObserver.h +++ b/qpid/cpp/src/qpid/broker/ConnectionObserver.h @@ -37,14 +37,19 @@ class ConnectionObserver public: virtual ~ConnectionObserver() {} - /** Called when a connection is opened and authentication has been - * performed. + /** Called when a connection is first established. */ + virtual void connection(Connection&) {} + + /** Called when the opening negotiation is done and the connection is authenticated. * @exception Throwing an exception will abort the connection. */ - virtual void connect(Connection& connection) = 0; + virtual void opened(Connection&) {} + + /** Called when a connection is closed. */ + virtual void closed(Connection&) {} - /** Called when a connection is torn down. */ - virtual void disconnect(Connection& connection) = 0; + /** Called when a connection is forced closed. */ + virtual void forced(Connection&, const std::string& /*message*/) {} }; }} // namespace qpid::broker diff --git a/qpid/cpp/src/qpid/broker/ConnectionObservers.h b/qpid/cpp/src/qpid/broker/ConnectionObservers.h index bef40e26b4..b36097cdbb 100644 --- a/qpid/cpp/src/qpid/broker/ConnectionObservers.h +++ b/qpid/cpp/src/qpid/broker/ConnectionObservers.h @@ -38,17 +38,29 @@ class ConnectionObservers : public ConnectionObserver { observers.push_back(observer); } - // implementation of ConnectionObserver interface - void connect(Connection& c) { - std::for_each(observers.begin(), observers.end(), boost::bind(&ConnectionObserver::connect, _1, boost::ref(c))); + void connection(Connection& c) { + each(boost::bind(&ConnectionObserver::connection, _1, boost::ref(c))); } - void disconnect(Connection& c) { - std::for_each(observers.begin(), observers.end(), boost::bind(&ConnectionObserver::disconnect, _1, boost::ref(c))); + + void opened(Connection& c) { + each(boost::bind(&ConnectionObserver::opened, _1, boost::ref(c))); + } + + void closed(Connection& c) { + each(boost::bind(&ConnectionObserver::closed, _1, boost::ref(c))); + } + + void forced(Connection& c, const std::string& text) { + each(boost::bind(&ConnectionObserver::forced, _1, boost::ref(c), text)); } private: typedef std::vector<boost::shared_ptr<ConnectionObserver> > Observers; Observers observers; + + template <class F> void each(F f) { + std::for_each(observers.begin(), observers.end(), f); + } }; }} // namespace qpid::broker diff --git a/qpid/cpp/src/qpid/broker/LinkRegistry.cpp b/qpid/cpp/src/qpid/broker/LinkRegistry.cpp index 31b4f1b490..1749440d7b 100644 --- a/qpid/cpp/src/qpid/broker/LinkRegistry.cpp +++ b/qpid/cpp/src/qpid/broker/LinkRegistry.cpp @@ -48,6 +48,16 @@ LinkRegistry::LinkRegistry () : { } +namespace { +struct ConnectionObserverImpl : public ConnectionObserver { + LinkRegistry& links; + ConnectionObserverImpl(LinkRegistry& l) : links(l) {} + void connection(Connection& c) { links.notifyConnection(c.getMgmtId(), &c); } + void closed(Connection& c) { links.notifyClosed(c.getMgmtId()); } + void forced(Connection& c, const string& text) { links.notifyConnectionForced(c.getMgmtId(), text); } +}; +} + LinkRegistry::LinkRegistry (Broker* _broker) : broker(_broker), timer(&broker->getTimer()), maintenanceTask(new Periodic(*this)), @@ -55,6 +65,8 @@ LinkRegistry::LinkRegistry (Broker* _broker) : realm(broker->getOptions().realm) { timer->add(maintenanceTask); + broker->getConnectionObservers().add( + boost::shared_ptr<ConnectionObserver>(new ConnectionObserverImpl(*this))); } LinkRegistry::~LinkRegistry() diff --git a/qpid/cpp/src/qpid/ha/ConnectionExcluder.h b/qpid/cpp/src/qpid/ha/ConnectionExcluder.h index 4a878c81c3..2ea75efffd 100644 --- a/qpid/cpp/src/qpid/ha/ConnectionExcluder.h +++ b/qpid/cpp/src/qpid/ha/ConnectionExcluder.h @@ -42,7 +42,7 @@ class ConnectionExcluder : public broker::ConnectionObserver ConnectionExcluder(string adminUser_, PrimaryTest isPrimary_) : adminUser(adminUser_), isPrimary(isPrimary_) {} - void connect(broker::Connection& connection) { + void opened(broker::Connection& connection) { if (!isPrimary() && !connection.isLink() && !connection.isAuthenticatedUser(adminUser)) { @@ -59,8 +59,6 @@ class ConnectionExcluder : public broker::ConnectionObserver } } - void disconnect(broker::Connection&) {} - private: string adminUser; PrimaryTest isPrimary; |