You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2012/06/08 17:24:31 UTC
svn commit: r1348114 - in /qpid/trunk/qpid/cpp/src: ./ qpid/broker/ qpid/ha/
Author: aconway
Date: Fri Jun 8 15:24:30 2012
New Revision: 1348114
URL: http://svn.apache.org/viewvc?rev=1348114&view=rev
Log:
QPID-3603: Add ConfigurationObserver.
Allows plugins to observe configuration events queue create/destroy,
exchange create/destroy and bind/unbind.
Added:
qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObserver.h (with props)
qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObservers.h (with props)
qpid/trunk/qpid/cpp/src/qpid/broker/Observers.h (with props)
Modified:
qpid/trunk/qpid/cpp/src/Makefile.am
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionObservers.h
qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
Modified: qpid/trunk/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/Makefile.am?rev=1348114&r1=1348113&r2=1348114&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/Makefile.am Fri Jun 8 15:24:30 2012
@@ -549,6 +549,8 @@ libqpidbroker_la_SOURCES = \
qpid/broker/ConsumerFactory.h \
qpid/broker/ConnectionObserver.h \
qpid/broker/ConnectionObservers.h \
+ qpid/broker/ConfigurationObserver.h \
+ qpid/broker/ConfigurationObservers.h \
qpid/broker/Daemon.cpp \
qpid/broker/Daemon.h \
qpid/broker/Deliverable.h \
@@ -611,6 +613,7 @@ libqpidbroker_la_SOURCES = \
qpid/broker/NameGenerator.h \
qpid/broker/NullMessageStore.cpp \
qpid/broker/NullMessageStore.h \
+ qpid/broker/Observers.h \
qpid/broker/OwnershipToken.h \
qpid/broker/Persistable.h \
qpid/broker/PersistableConfig.h \
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp?rev=1348114&r1=1348113&r2=1348114&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Fri Jun 8 15:24:30 2012
@@ -1219,6 +1219,7 @@ void Broker::bind(const std::string& que
throw framing::NotFoundException(QPID_MSG("Bind failed. No such exchange: " << exchangeName));
} else {
if (queue->bind(exchange, key, arguments)) {
+ getConfigurationObservers().bind(exchange, queue, key, arguments);
if (managementAgent.get()) {
managementAgent->raiseEvent(_qmf::EventBind(connectionId, userId, exchangeName,
queueName, key, ManagementAgent::toMap(arguments)));
@@ -1254,6 +1255,8 @@ void Broker::unbind(const std::string& q
if (exchange->isDurable() && queue->isDurable()) {
store->unbind(*exchange, *queue, key, qpid::framing::FieldTable());
}
+ getConfigurationObservers().unbind(
+ exchange, queue, key, framing::FieldTable());
if (managementAgent.get()) {
managementAgent->raiseEvent(_qmf::EventUnbind(connectionId, userId, exchangeName, queueName, key));
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h?rev=1348114&r1=1348113&r2=1348114&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h Fri Jun 8 15:24:30 2012
@@ -39,6 +39,7 @@
#include "qpid/broker/ExpiryPolicy.h"
#include "qpid/broker/ConsumerFactory.h"
#include "qpid/broker/ConnectionObservers.h"
+#include "qpid/broker/ConfigurationObservers.h"
#include "qpid/management/Manageable.h"
#include "qpid/management/ManagementAgent.h"
#include "qmf/org/apache/qpid/broker/Broker.h"
@@ -184,6 +185,7 @@ class Broker : public sys::Runnable, pub
AclModule* acl;
DataDir dataDir;
ConnectionObservers connectionObservers;
+ ConfigurationObservers configurationObservers;
QueueRegistry queues;
ExchangeRegistry exchanges;
@@ -210,7 +212,6 @@ class Broker : public sys::Runnable, pub
mutable sys::Mutex linkClientPropertiesLock;
framing::FieldTable linkClientProperties;
-
public:
QPID_BROKER_EXTERN virtual ~Broker();
@@ -381,6 +382,7 @@ class Broker : public sys::Runnable, pub
ConsumerFactories& getConsumerFactories() { return consumerFactories; }
ConnectionObservers& getConnectionObservers() { return connectionObservers; }
+ ConfigurationObservers& getConfigurationObservers() { return configurationObservers; }
/** Properties to be set on outgoing link connections */
QPID_BROKER_EXTERN framing::FieldTable getLinkClientProperties() const;
Added: qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObserver.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObserver.h?rev=1348114&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObserver.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObserver.h Fri Jun 8 15:24:30 2012
@@ -0,0 +1,61 @@
+#ifndef QPID_BROKER_CONFIGURATIONOBSERVER_H
+#define QPID_BROKER_CONFIGURATIONOBSERVER_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 <boost/shared_ptr.hpp>
+#include <string>
+
+namespace qpid {
+
+namespace framing {
+class FieldTable;
+}
+
+namespace broker {
+class Queue;
+class Exchange;
+
+
+/**
+ * Observer for changes to configuration (aka wiring)
+ */
+class ConfigurationObserver
+{
+ public:
+ virtual ~ConfigurationObserver() {}
+ virtual void queueCreate(const boost::shared_ptr<Queue>&) {}
+ virtual void queueDestroy(const boost::shared_ptr<Queue>&) {}
+ virtual void exchangeCreate(const boost::shared_ptr<Exchange>&) {}
+ virtual void exchangeDestroy(const boost::shared_ptr<Exchange>&) {}
+ virtual void bind(const boost::shared_ptr<Exchange>& ,
+ const boost::shared_ptr<Queue>& ,
+ const std::string& /*key*/,
+ const framing::FieldTable& /*args*/) {}
+ virtual void unbind(const boost::shared_ptr<Exchange>&,
+ const boost::shared_ptr<Queue>& ,
+ const std::string& /*key*/,
+ const framing::FieldTable& /*args*/) {}
+};
+}} // namespace qpid::broker
+
+#endif /*!QPID_BROKER_CONFIGURATIONOBSERVER_H*/
Propchange: qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObserver.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObserver.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObservers.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObservers.h?rev=1348114&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObservers.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObservers.h Fri Jun 8 15:24:30 2012
@@ -0,0 +1,72 @@
+#ifndef QPID_BROKER_CONFIGURATIONOBSERVERS_H
+#define QPID_BROKER_CONFIGURATIONOBSERVERS_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 "ConfigurationObserver.h"
+#include "Observers.h"
+#include "qpid/sys/Mutex.h"
+
+namespace qpid {
+namespace broker {
+
+/**
+ * A configuration observer that delegates to a collection of
+ * configuration observers.
+ *
+ * THREAD SAFE
+ */
+class ConfigurationObservers : public ConfigurationObserver,
+ public Observers<ConfigurationObserver>
+{
+ public:
+ void queueCreate(const boost::shared_ptr<Queue>& q) {
+ each(boost::bind(&ConfigurationObserver::queueCreate, _1, q));
+ }
+ void queueDestroy(const boost::shared_ptr<Queue>& q) {
+ each(boost::bind(&ConfigurationObserver::queueDestroy, _1, q));
+ }
+ void exchangeCreate(const boost::shared_ptr<Exchange>& e) {
+ each(boost::bind(&ConfigurationObserver::exchangeCreate, _1, e));
+ }
+ void exchangeDestroy(const boost::shared_ptr<Exchange>& e) {
+ each(boost::bind(&ConfigurationObserver::exchangeDestroy, _1, e));
+ }
+ void bind(const boost::shared_ptr<Exchange>& exchange,
+ const boost::shared_ptr<Queue>& queue,
+ const std::string& key,
+ const framing::FieldTable& args) {
+ each(boost::bind(
+ &ConfigurationObserver::bind, _1, exchange, queue, key, args));
+ }
+ void unbind(const boost::shared_ptr<Exchange>& exchange,
+ const boost::shared_ptr<Queue>& queue,
+ const std::string& key,
+ const framing::FieldTable& args) {
+ each(boost::bind(
+ &ConfigurationObserver::unbind, _1, exchange, queue, key, args));
+ }
+};
+
+}} // namespace qpid::broker
+
+#endif /*!QPID_BROKER_CONFIGURATIONOBSERVERS_H*/
Propchange: qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObservers.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/cpp/src/qpid/broker/ConfigurationObservers.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionObservers.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionObservers.h?rev=1348114&r1=1348113&r2=1348114&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionObservers.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionObservers.h Fri Jun 8 15:24:30 2012
@@ -23,9 +23,7 @@
*/
#include "ConnectionObserver.h"
-#include "qpid/sys/Mutex.h"
-#include <set>
-#include <algorithm>
+#include "Observers.h"
namespace qpid {
namespace broker {
@@ -35,18 +33,10 @@ namespace broker {
* Calling a ConnectionObserver function will call that function on each observer.
* THREAD SAFE.
*/
-class ConnectionObservers : public ConnectionObserver {
+class ConnectionObservers : public ConnectionObserver,
+ public Observers<ConnectionObserver>
+{
public:
- void add(boost::shared_ptr<ConnectionObserver> observer) {
- sys::Mutex::ScopedLock l(lock);
- observers.insert(observer);
- }
-
- void remove(boost::shared_ptr<ConnectionObserver> observer) {
- sys::Mutex::ScopedLock l(lock);
- observers.erase(observer);
- }
-
void connection(Connection& c) {
each(boost::bind(&ConnectionObserver::connection, _1, boost::ref(c)));
}
@@ -62,16 +52,6 @@ class ConnectionObservers : public Conne
void forced(Connection& c, const std::string& text) {
each(boost::bind(&ConnectionObserver::forced, _1, boost::ref(c), text));
}
-
- private:
- typedef std::set<boost::shared_ptr<ConnectionObserver> > Observers;
- sys::Mutex lock;
- Observers observers;
-
- template <class F> void each(F f) {
- sys::Mutex::ScopedLock l(lock);
- std::for_each(observers.begin(), observers.end(), f);
- }
};
}} // namespace qpid::broker
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp?rev=1348114&r1=1348113&r2=1348114&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp Fri Jun 8 15:24:30 2012
@@ -19,6 +19,7 @@
*
*/
+#include "qpid/broker/Broker.h"
#include "qpid/broker/ExchangeRegistry.h"
#include "qpid/broker/DirectExchange.h"
#include "qpid/broker/FanOutExchange.h"
@@ -69,6 +70,7 @@ pair<Exchange::shared_ptr, bool> Exchang
exchange = i->second(name, durable, args, parent, broker);
}
}
+ if (broker) broker->getConfigurationObservers().exchangeCreate(exchange);
exchanges[name] = exchange;
return std::pair<Exchange::shared_ptr, bool>(exchange, true);
} else {
@@ -85,8 +87,10 @@ void ExchangeRegistry::destroy(const str
RWlock::ScopedWlock locker(lock);
ExchangeMap::iterator i = exchanges.find(name);
if (i != exchanges.end()) {
+ Exchange::shared_ptr ex = i->second;
i->second->destroy();
exchanges.erase(i);
+ if (broker) broker->getConfigurationObservers().exchangeDestroy(ex);
}
}
Added: qpid/trunk/qpid/cpp/src/qpid/broker/Observers.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Observers.h?rev=1348114&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Observers.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Observers.h Fri Jun 8 15:24:30 2012
@@ -0,0 +1,69 @@
+#ifndef QPID_BROKER_OBSERVERS_H
+#define QPID_BROKER_OBSERVERS_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/sys/Mutex.h"
+#include <boost/shared_ptr.hpp>
+#include <vector>
+#include <algorithm>
+
+namespace qpid {
+namespace broker {
+
+/**
+ * Base class for collections of observers with thread-safe add/remove and traversal.
+ */
+template <class Observer>
+class Observers
+{
+ public:
+ void add(boost::shared_ptr<Observer> observer) {
+ sys::Mutex::ScopedLock l(lock);
+ observers.push_back(observer);
+ }
+
+ void remove(boost::shared_ptr<Observer> observer) {
+ sys::Mutex::ScopedLock l(lock);
+ typename List::iterator i = std::find(observers.begin(), observers.end(), observer);
+ observers.erase(i);
+ }
+
+ protected:
+ typedef std::vector<boost::shared_ptr<Observer> > List;
+
+ sys::Mutex lock;
+ List observers;
+
+ template <class F> void each(F f) {
+ List copy;
+ {
+ sys::Mutex::ScopedLock l(lock);
+ copy = observers;
+ }
+ std::for_each(copy.begin(), copy.end(), f);
+ }
+};
+
+}} // namespace qpid::broker
+
+#endif /*!QPID_BROKER_OBSERVERS_H*/
Propchange: qpid/trunk/qpid/cpp/src/qpid/broker/Observers.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/cpp/src/qpid/broker/Observers.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp?rev=1348114&r1=1348113&r2=1348114&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp Fri Jun 8 15:24:30 2012
@@ -18,6 +18,7 @@
* under the License.
*
*/
+#include "qpid/broker/Broker.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/QueueRegistry.h"
#include "qpid/broker/QueueEvents.h"
@@ -64,17 +65,22 @@ QueueRegistry::declare(const string& dec
//i.e. recovering a queue for which we already have a persistent record
queue->configure(arguments);
}
+ if (broker) broker->getConfigurationObservers().queueCreate(queue);
queues[name] = queue;
if (lastNode) queue->setLastNodeFailure();
-
return std::pair<Queue::shared_ptr, bool>(queue, true);
} else {
return std::pair<Queue::shared_ptr, bool>(i->second, false);
}
}
-void QueueRegistry::destroyLH (const string& name){
- queues.erase(name);
+void QueueRegistry::destroyLH (const string& name) {
+ QueueMap::iterator i = queues.find(name);
+ if (i != queues.end()) {
+ Queue::shared_ptr q = i->second;
+ queues.erase(i);
+ if (broker) broker->getConfigurationObservers().queueDestroy(q);
+ }
}
void QueueRegistry::destroy (const string& name){
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp?rev=1348114&r1=1348113&r2=1348114&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp Fri Jun 8 15:24:30 2012
@@ -94,10 +94,11 @@ HaBroker::HaBroker(broker::Broker& b, co
if (!settings.brokerUrl.empty()) setBrokerUrl(Url(settings.brokerUrl), l);
statusChanged(l);
- QPID_LOG(notice, logPrefix << "Broker starting on " << brokerInfo);
+ QPID_LOG(notice, logPrefix << "Broker starting: " << brokerInfo);
}
HaBroker::~HaBroker() {
+ QPID_LOG(debug, logPrefix << "Broker shut down: " << brokerInfo);
broker.getConnectionObservers().remove(excluder);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org