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 2007/06/29 19:59:07 UTC
svn commit: r551981 - in /incubator/qpid/trunk/qpid/cpp/src: ./ qpid/
qpid/broker/ qpid/cluster/ qpid/framing/ tests/
Author: aconway
Date: Fri Jun 29 10:59:00 2007
New Revision: 551981
URL: http://svn.apache.org/viewvc?view=rev&rev=551981
Log:
* Summary:
- Improved plugin framework and HandlerUpdater interface.
- Cluster handlers for traffic to/from cluster.
- Cluster HandlerUpdater configures channel chains for cluster.
- Cluster PluginProvider registers cluster objects with broker.
* src/qpid/framing/AMQFrame.h: Made data members public. Handlers
need to be able to modify frame data, getters/setters are just a
nuisance here.
* src/tests/Cluster.cpp: Updated for cluster changes, using
handlers instead of friendship to hook test into Cluster code.
* src/qpid/framing/amqp_types.h: Added CHANNEL_MAX and
CHANNEL_HIGH_BIT constants.
* src/qpid/framing/HandlerUpdater.h: Renamed ChannelInitializer,
broke dependency on broker channel types.
* src/qpid/framing/Handler.h: Added constructors and nextHandler()
* src/qpid/framing/AMQFrame.h (class AMQFrame): Inlined getChannel()
* src/qpid/cluster/ClusterPluginProvider.cpp: Provider for cluster
plugins.
* src/qpid/cluster/Cluster.cpp: Use ChannelManager. Factor out
plugin details to ClusterPluginProvider.
* src/qpid/cluster/ChannelManager.h: Insert cluster handlers
into channel chains, route frames between cluster and channels.
* src/qpid/broker/BrokerAdapter.cpp (startOk): use CHANNEL_MAX
constant.
* src/qpid/broker/Broker.cpp:
- Refactored for new plugin framework.
- Added getUrl().
* src/qpid/Url.h: Added constructor from Address.
* src/qpid/Plugin.h: Generalized plugin framework, broke
dependency on Broker interfaces. We may want to use plug-ins for
clients also at some point.
* src/tests/run_test: Fix bug when VALGRIND is not set.
Added:
incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.cpp
- copied, changed from r551363, incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h
- copied, changed from r551363, incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.h
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.cpp (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.h (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPluginProvider.cpp (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/HandlerUpdater.h
- copied, changed from r551363, incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ChannelInitializer.h
Removed:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ChannelInitializer.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.h
Modified:
incubator/qpid/trunk/qpid/cpp/src/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/cluster.mk
incubator/qpid/trunk/qpid/cpp/src/qpid/Url.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/Url.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.h
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cpg.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h
incubator/qpid/trunk/qpid/cpp/src/qpidd.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/.valgrindrc-default
incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h
incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/InProcessBroker.h
incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/tests/Url.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/run_test
Modified: incubator/qpid/trunk/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/Makefile.am?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/Makefile.am Fri Jun 29 10:59:00 2007
@@ -152,11 +152,14 @@
qpid/framing/Proxy.cpp \
qpid/framing/Handler.h \
qpid/framing/FrameHandler.h \
+ qpid/framing/HandlerUpdater.h \
gen/qpid/framing/AMQP_ClientProxy.cpp \
gen/qpid/framing/AMQP_HighestVersion.h \
gen/qpid/framing/AMQP_MethodVersionMap.cpp \
gen/qpid/framing/AMQP_ServerProxy.cpp \
qpid/Exception.cpp \
+ qpid/Plugin.h \
+ qpid/Plugin.cpp \
qpid/Url.h \
qpid/Url.cpp \
qpid/QpidError.cpp \
@@ -217,10 +220,7 @@
qpid/broker/TopicExchange.cpp \
qpid/broker/TxAck.cpp \
qpid/broker/TxBuffer.cpp \
- qpid/broker/TxPublish.cpp \
- qpid/broker/Plugin.h \
- qpid/broker/Plugin.cpp \
- qpid/broker/ChannelInitializer.h
+ qpid/broker/TxPublish.cpp
libqpidclient_la_LIBADD = libqpidcommon.la
libqpidclient_la_SOURCES = \
Modified: incubator/qpid/trunk/qpid/cpp/src/cluster.mk
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/cluster.mk?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/cluster.mk (original)
+++ incubator/qpid/trunk/qpid/cpp/src/cluster.mk Fri Jun 29 10:59:00 2007
@@ -10,9 +10,12 @@
qpid/cluster/Cluster.h \
qpid/cluster/Cpg.cpp \
qpid/cluster/Cpg.h \
- qpid/cluster/Dispatchable.h
+ qpid/cluster/Dispatchable.h \
+ qpid/cluster/ChannelManager.h \
+ qpid/cluster/ChannelManager.cpp \
+ qpid/cluster/ClusterPluginProvider.cpp
-libqpidcluster_la_LIBADD= -lcpg libqpidcommon.la
+libqpidcluster_la_LIBADD= -lcpg libqpidbroker.la
else
# Empty stub library to satisfy rpm spec file.
Copied: incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.cpp (from r551363, incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.cpp)
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.cpp?view=diff&rev=551981&p1=incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.cpp&r1=551363&p2=incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.cpp&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.cpp Fri Jun 29 10:59:00 2007
@@ -19,24 +19,24 @@
#include "Plugin.h"
namespace qpid {
-namespace broker {
-std::vector<Plugin*> Plugin::plugins;
+std::vector<PluginProvider*> PluginProvider::providers;
-Plugin::Plugin() {
+PluginProvider::PluginProvider() {
// Register myself.
- plugins.push_back(this);
+ providers.push_back(this);
}
-Plugin::~Plugin() {}
+PluginProvider::~PluginProvider() {}
-Options* Plugin::getOptions() { return 0; }
+Options* PluginProvider::getOptions() { return 0; }
-void Plugin::start(Broker&) {}
-
-void Plugin::finish(Broker&) {}
+const std::vector<PluginProvider*>& PluginProvider::getProviders() {
+ return providers;
+}
+} // namespace qpid
-const std::vector<Plugin*>& Plugin::getPlugins() { return plugins; }
+// TODO aconway 2007-06-28: GNU lib has portable dlopen if we go that way.
#ifdef USE_APR_PLATFORM
@@ -44,24 +44,27 @@
#include "qpid/sys/apr/APRPool.h"
#include <apr_dso.h>
-void Plugin::dlopen(const std::string& name) {
+namespace qpid {
+void dlopen(const char* name) {
apr_dso_handle_t* handle;
CHECK_APR_SUCCESS(
- apr_dso_load(&handle, name.c_str(), sys::APRPool::get()));
+ apr_dso_load(&handle, name, sys::APRPool::get()));
}
+} // namespace qpid
#else // Posix
#include <dlfcn.h>
-void Plugin::dlopen(const std::string& name) {
+namespace qpid {
+void dlopen(const char* name) {
dlerror();
- dlopen(name.c_str(), RTLD_NOW);
+ dlopen(name, RTLD_NOW);
const char* error = dlerror();
if (error) {
THROW_QPID_ERROR(INTERNAL_ERROR, error);
}
}
-#endif // USE_APR_PLATFORM
+} // namespace qpidpp
-}} // namespace qpid::broker
+#endif // USE_APR_PLATFORM
Copied: incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h (from r551363, incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.h)
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h?view=diff&rev=551981&p1=incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.h&r1=551363&p2=incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Plugin.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h Fri Jun 29 10:59:00 2007
@@ -1,5 +1,5 @@
-#ifndef QPID_BROKER_PLUGIN_H
-#define QPID_BROKER_PLUGIN_H
+#ifndef QPID_PLUGIN_H
+#define QPID_PLUGIN_H
/*
*
@@ -19,62 +19,93 @@
*
*/
-#include "qpid/Options.h"
-#include "qpid/broker/Broker.h"
+#include "qpid/shared_ptr.h"
#include <boost/noncopyable.hpp>
#include <vector>
+#include <boost/function.hpp>
+
+
+/**@file Generic plug-in framework. */
namespace qpid {
-namespace broker {
+class Options;
+
+/** Generic base class to allow dynamic casting of generic Plugin objects
+ * to concrete types.
+ */
+struct Plugin : private boost::noncopyable {
+ virtual ~Plugin() {}
+};
+
+/** Generic interface for anything that uses plug-ins. */
+struct PluginUser : boost::noncopyable {
+ virtual ~PluginUser() {}
+ /**
+ * Called by a PluginProvider to provide a plugin.
+ *
+ * A concrete PluginUser will dynamic_pointer_cast plugin to a
+ * class it knows how to use. A PluginUser should ignore plugins
+ * it does not recognize.
+ *
+ * The user will release its shared_ptr when it is finished using
+ * plugin.
+ */
+ virtual void use(const shared_ptr<Plugin>& plugin) = 0;
+};
+
/**
- * Inherit broker plug-ins from this class, override the virtual
- * functions and create a global (or class static member) instance in
- * a shared library. When the library is loaded your plug-in will be
- * registered.
+ * Base for classes that provide plug-ins.
*/
-class Plugin : boost::noncopyable
+class PluginProvider : boost::noncopyable
{
public:
- /** Constructor registers the plugin to appear in getPlugins().
- * Note: Plugin subclasses should only be constructed during
- * static initialization, i.e. they should only be declared
- * as global or static member variables.
+ /**
+ * Register the provider to appear in getProviders()
+ *
+ * A concrete PluginProvider is instantiated as a global or static
+ * member variable in a library so it is registered during static
+ * initialization when the library is loaded.
*/
- Plugin();
+ PluginProvider();
- virtual ~Plugin();
+ virtual ~PluginProvider();
/**
- * Override if your plugin has configuration options.
- * They will be included in options parsing prior to broker
- * creation setup.
- *@return An options group or 0. Default returns 0.
+ * Returns configuration options for the plugin.
+ * Then will be updated during option parsing by the host program.
+ *
+ * @return An options group or 0 for no options. Default returns 0.
+ * PluginProvider retains ownership of return value.
*/
virtual Options* getOptions();
- /** Called immediately after broker creation to allow plug-ins
- * to do whatever they do to the broker, e.g. add handler chain
- * manipulators.
+ /** Provide plugins to a PluginUser.
+ *
+ * The provider can dynamic_cast the user if it only provides
+ * plugins to certain types of user. Providers should ignore
+ * users they don't recognize.
*/
- virtual void start(Broker& b);
-
- /** Called just before broker shutdown. Default does nothing */
- virtual void finish(Broker& b);
+ virtual void provide(PluginUser& user) = 0;
- /** Get the list of registered plug-ins. */
- static const std::vector<Plugin*>& getPlugins();
+ /** Get the list of pointers to the registered providers.
+ * Caller must not delete the pointers.
+ */
+ static const std::vector<PluginProvider*>& getProviders();
- /** Load a shared library (that contains plugins presumably!) */
- static void dlopen(const std::string& libname);
-
private:
- static std::vector<Plugin*> plugins;
+ static std::vector<PluginProvider*> providers;
};
-
-}} // namespace qpid::broker
-
-
+
+/** Load a shared library, registering any PluginProvider it contains.
+ *
+ * This is just a convenient portable wrapper for normal shared
+ * library loading. A global PluginProvider instance loaded or
+ * linked in any way will get registered.
+ */
+void dlopen(const char* libname);
+
+} // namespace qpid
-#endif /*!QPID_BROKER_PLUGIN_H*/
+#endif /*!QPID_PLUGIN_H*/
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/Url.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/Url.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/Url.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/Url.cpp Fri Jun 29 10:59:00 2007
@@ -100,14 +100,13 @@
};
};
-Url::Url(const string& url) {
- if (!parse(url.c_str(), UrlGrammar(*this)).full)
- throw InvalidUrl("Invalid AMQP url: "+url);
- // TODO aconway 2007-06-15: Better error handling?
+void Url::parse(const char* url) {
+ if (!boost::spirit::parse(url, UrlGrammar(*this)).full)
+ throw InvalidUrl(string("Invalid AMQP url: ")+url);
}
-Url::Url(const string& url, const nothrow_t&) {
- if (!parse(url.c_str(), UrlGrammar(*this)).full)
+void Url::parseNoThrow(const char* url) {
+ if (!boost::spirit::parse(url, UrlGrammar(*this)).full)
clear();
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/Url.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/Url.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/Url.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/Url.h Fri Jun 29 10:59:00 2007
@@ -31,7 +31,7 @@
/** TCP address of a broker - host:port */
struct TcpAddress {
static const uint16_t DEFAULT_PORT=5672;
- TcpAddress(const std::string& host_=std::string(),
+ explicit TcpAddress(const std::string& host_=std::string(),
uint16_t port_=DEFAULT_PORT)
: host(host_), port(port_) {}
std::string host;
@@ -62,17 +62,26 @@
/** Empty URL. */
Url() {}
- /** Parse an amqp URL string as defined in
+ /** URL containing a single address */
+ explicit Url(const Address& addr) { push_back(addr); }
+
+ /** Parse url, throw InvalidUrl if invalid. */
+ explicit Url(const std::string& url) { parse(url.c_str()); }
+
+ /** Parse url, throw InvalidUrl if invalid. */
+ explicit Url(const char* url) { parse(url); }
+
+ /** Replace contents with parsed URL as defined in
* https://wiki.108.redhat.com/jira/browse/AMQP-95
*@exception InvalidUrl if the url is invalid.
*/
- Url(const std::string& url);
+ void parse(const char* url);
- /** Parse an amqp URL string as defined in
+ /** Replace contesnts with parsed URL as defined in
* https://wiki.108.redhat.com/jira/browse/AMQP-95
- * Results in empty URL if url is invalid.
+ * url.empty() will be true if url is invalid.
*/
- Url(const std::string& url, const std::nothrow_t&);
+ void parseNoThrow(const char* url);
};
std::ostream& operator<<(std::ostream& os, const Url& url);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Fri Jun 29 10:59:00 2007
@@ -22,6 +22,7 @@
#include "Broker.h"
#include "qpid/framing/AMQFrame.h"
+#include "qpid/framing/HandlerUpdater.h"
#include "DirectExchange.h"
#include "TopicExchange.h"
#include "FanOutExchange.h"
@@ -41,6 +42,7 @@
#include <memory>
using qpid::sys::Acceptor;
+using qpid::framing::HandlerUpdater;
namespace qpid {
namespace broker {
@@ -100,16 +102,17 @@
}
-Broker::shared_ptr Broker::create(int16_t port)
+shared_ptr<Broker> Broker::create(int16_t port)
{
Options config;
config.port=port;
return create(config);
}
-Broker::shared_ptr Broker::create(const Options& config) {
- return Broker::shared_ptr(new Broker(config));
-}
+shared_ptr<Broker> Broker::create(const Options& opts)
+{
+ return shared_ptr<Broker>(new Broker(opts));
+}
MessageStore* Broker::createStore(const Options& config) {
if (config.store.empty())
@@ -134,6 +137,10 @@
int16_t Broker::getPort() const { return getAcceptor().getPort(); }
+std::string Broker::getUrl() const {
+ return Url(TcpAddress(getAcceptor().getHost(), getPort())).str();
+}
+
Acceptor& Broker::getAcceptor() const {
if (!acceptor)
const_cast<Acceptor::shared_ptr&>(acceptor) =
@@ -144,6 +151,14 @@
return *acceptor;
}
+void Broker::use(const shared_ptr<Plugin>& plugin) {
+ shared_ptr<HandlerUpdater> updater=
+ dynamic_pointer_cast<HandlerUpdater>(plugin);
+ if (updater) {
+ QPID_LOG(critical, "HandlerUpdater plugins not implemented");
+ // FIXME aconway 2007-06-28: hook into Connections.
+ }
+}
}} // namespace qpid::broker
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h Fri Jun 29 10:59:00 2007
@@ -24,9 +24,9 @@
#include "ConnectionFactory.h"
#include "qpid/Url.h"
+#include "qpid/Plugin.h"
#include "qpid/sys/Runnable.h"
#include "qpid/sys/Acceptor.h"
-#include "qpid/SharedObject.h"
#include "MessageStore.h"
#include "AutoDelete.h"
#include "ExchangeRegistry.h"
@@ -39,14 +39,17 @@
#include "qpid/Options.h"
namespace qpid {
+
+namespace framing {
+class HandlerUpdater;
+}
+
namespace broker {
-class ChannelInitializer;
/**
* A broker instance.
*/
-class Broker : public sys::Runnable,
- public SharedObject<Broker>
+class Broker : public sys::Runnable, public PluginUser
{
public:
struct Options : public qpid::Options {
@@ -62,16 +65,9 @@
virtual ~Broker();
- /**
- * Create a broker.
- * @param port Port to listen on or 0 to pick a port dynamically.
- */
- static shared_ptr create(int16_t port = TcpAddress::DEFAULT_PORT);
-
- /**
- * Create a broker with the options in config.
- */
- static shared_ptr create(const Options& config);
+ Broker(const Options& configuration);
+ static shared_ptr<Broker> create(const Options& configuration);
+ static shared_ptr<Broker> create(int16_t port = TcpAddress::DEFAULT_PORT);
/**
* Return listening port. If called before bind this is
@@ -80,19 +76,22 @@
* 0.
*/
virtual int16_t getPort() const;
-
+
+ /** Return the broker's URL. */
+ virtual std::string getUrl() const;
+
/**
* Run the broker. Implements Runnable::run() so the broker
* can be run in a separate thread.
*/
virtual void run();
- /** Plug-in a channel initializer. */
- void plugin(const qpid::shared_ptr<ChannelInitializer>&);
-
/** Shut down the broker */
virtual void shutdown();
+ /** Use a plugin */
+ void use(const shared_ptr<Plugin>& plugin);
+
MessageStore& getStore() { return *store; }
QueueRegistry& getQueues() { return queues; }
ExchangeRegistry& getExchanges() { return exchanges; }
@@ -102,7 +101,6 @@
DtxManager& getDtxManager() { return dtxManager; }
private:
- Broker(const Options& configuration);
sys::Acceptor& getAcceptor() const;
Options config;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp Fri Jun 29 10:59:00 2007
@@ -58,7 +58,7 @@
const string& /*response*/, const string& /*locale*/)
{
client.tune(
- 100, connection.getFrameMax(), connection.getHeartbeat());
+ CHANNEL_MAX, connection.getFrameMax(), connection.getHeartbeat());
}
void BrokerAdapter::ConnectionHandlerImpl::secureOk(
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.cpp Fri Jun 29 10:59:00 2007
@@ -24,13 +24,13 @@
BrokerSingleton::BrokerSingleton() {
if (broker.get() == 0)
broker = Broker::create();
- Broker::shared_ptr::operator=(broker);
+ shared_ptr<Broker>::operator=(broker);
}
BrokerSingleton::~BrokerSingleton() {
broker->shutdown();
}
-Broker::shared_ptr BrokerSingleton::broker;
+shared_ptr<Broker> BrokerSingleton::broker;
}} // namespace qpid::broker
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerSingleton.h Fri Jun 29 10:59:00 2007
@@ -32,17 +32,17 @@
* Useful for unit tests that want to share a broker between multiple
* tests to reduce overhead of starting/stopping a broker for every test.
*
- * Tests that need a new broker can call Broker::create directly.
+ * Tests that need a new broker can create it directly.
*
* THREAD UNSAFE.
*/
-class BrokerSingleton : public Broker::shared_ptr
+class BrokerSingleton : public shared_ptr<Broker>
{
public:
BrokerSingleton();
~BrokerSingleton();
private:
- static Broker::shared_ptr broker;
+ static shared_ptr<Broker> broker;
};
}} // namespace qpid::broker
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.cpp?view=auto&rev=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.cpp Fri Jun 29 10:59:00 2007
@@ -0,0 +1,85 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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/log/Statement.h"
+#include "qpid/framing/amqp_types.h"
+#include "qpid/framing/AMQFrame.h"
+#include "ChannelManager.h"
+
+namespace qpid {
+namespace cluster {
+
+using namespace framing;
+
+/** Handler to multicast to the cluster */
+struct ClusterHandler : public FrameHandler {
+
+ ClusterHandler(FrameHandler::Chain next, ChannelId bitmask_)
+ : FrameHandler(next), bitmask(bitmask_) {}
+
+ void handle(AMQFrame& frame) {
+ frame.channel |= bitmask; // Mark the frame
+ nextHandler(frame);
+ // TODO aconway 2007-06-28: Right now everything is backed up
+ // via multicast. When we have point-to-point backups this
+ // function must determine where each frame should be sent: to
+ // multicast or only to specific backup(s) via AMQP.
+ }
+
+ ChannelId bitmask;
+};
+
+ChannelManager::ChannelManager(FrameHandler::Chain mcast) : mcastOut(mcast){}
+
+void ChannelManager::update(ChannelId id, FrameHandler::Chains& chains) {
+ // Store the original cluster chains for the channel.
+ channels[id] = chains;
+
+ // Replace chains with multicast-to-cluster handlers that mark the
+ // high-bit of the channel ID on outgoing frames so we can tell
+ // them from incoming frames in handle()
+ //
+ // When handle() receives the frames from the cluster it
+ // will forward them to the original channel chains stored in
+ // channels map.
+ //
+ chains.in = make_shared_ptr(new ClusterHandler(mcastOut, 0));
+ chains.out= make_shared_ptr(new ClusterHandler(mcastOut, CHANNEL_HIGH_BIT));
+}
+
+void ChannelManager::handle(AMQFrame& frame) {
+ bool isOut = frame.channel | CHANNEL_HIGH_BIT;
+ frame.channel |= ~CHANNEL_HIGH_BIT; // Clear the bit.
+ ChannelMap::iterator i = channels.find(frame.getChannel());
+ if (i != channels.end()) {
+ Chain& chain = isOut ? i->second.out : i->second.in;
+ chain->handle(frame);
+ }
+ else
+ updateFailoverState(frame);
+}
+
+void ChannelManager::updateFailoverState(AMQFrame& ) {
+ QPID_LOG(critical, "Failover is not implemented");
+ // FIXME aconway 2007-06-28:
+ // If the channel is not in my map then I'm not primary so
+ // I don't pass the frame to the channel handler but I
+ // do need to update the session failover state.
+}
+
+}} // namespace qpid::cluster
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.h?view=auto&rev=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.h Fri Jun 29 10:59:00 2007
@@ -0,0 +1,66 @@
+#ifndef QPID_CLUSTER_CHANNELMANAGER_H
+#define QPID_CLUSTER_CHANNELMANAGER_H
+
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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/framing/HandlerUpdater.h"
+#include <map>
+
+namespace qpid {
+namespace cluster {
+
+/**
+ * Manage channels and handler chains on channels for the cluster.
+ *
+ * As HandlerUpdater plugin, updates channel handler chains with
+ * cluster handlers.
+ *
+ * As a FrameHandler handles frames coming from the cluster and
+ * dispatches them to the appropriate channel handler.
+ *
+ */
+class ChannelManager : public framing::HandlerUpdater,
+ public framing::FrameHandler
+{
+ public:
+ /** Takes a handler to send frames to the cluster. */
+ ChannelManager(framing::FrameHandler::Chain mcastOut);
+
+ /** As FrameHandler handle frames from the cluster */
+ void handle(framing::AMQFrame& frame);
+
+ /** As ChannelUpdater update the handler chains. */
+ void update(framing::ChannelId id, framing::FrameHandler::Chains& chains);
+
+ private:
+ void updateFailoverState(framing::AMQFrame&);
+
+ typedef std::map<framing::ChannelId,
+ framing::FrameHandler::Chains> ChannelMap;
+
+ framing::FrameHandler::Chain mcastOut;
+ ChannelMap channels;
+};
+
+
+}} // namespace qpid::cluster
+
+
+
+#endif /*!QPID_CLUSTER_CHANNELMANAGER_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ChannelManager.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Fri Jun 29 10:59:00 2007
@@ -17,13 +17,13 @@
*/
#include "Cluster.h"
-#include "Cpg.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/framing/ClusterNotifyBody.h"
#include "qpid/log/Statement.h"
#include <boost/bind.hpp>
#include <algorithm>
#include <iterator>
+#include <map>
namespace qpid {
namespace cluster {
@@ -35,40 +35,62 @@
return out << cluster.name.str() << "(" << cluster.self << ")";
}
+namespace {
+Cluster::Member::Status statusMap[CPG_REASON_PROCDOWN+1];
+struct StatusMapInit {
+ StatusMapInit() {
+ statusMap[CPG_REASON_JOIN] = Cluster::Member::JOIN;
+ statusMap[CPG_REASON_LEAVE] = Cluster::Member::LEAVE;
+ statusMap[CPG_REASON_NODEDOWN] = Cluster::Member::NODEDOWN;
+ statusMap[CPG_REASON_NODEUP] = Cluster::Member::NODEUP;
+ statusMap[CPG_REASON_PROCDOWN] = Cluster::Member::PROCDOWN;
+ }
+} instance;
+}
+
+Cluster::Member::Member(const cpg_address& addr)
+ : status(statusMap[addr.reason]) {}
+
void Cluster::notify() {
+ ProtocolVersion version;
// TODO aconway 2007-06-25: Use proxy here.
AMQFrame frame(version, 0,
make_shared_ptr(new ClusterNotifyBody(version, url)));
handle(frame);
}
-Cluster::Cluster(
- const std::string& name_, const std::string& url_, FrameHandler& next_,
- ProtocolVersion ver)
- : name(name_), url(url_), version(ver),
- cpg(new Cpg(boost::bind(&Cluster::cpgDeliver, this, _1, _2, _3, _4, _5, _6),
- boost::bind(&Cluster::cpgConfigChange, this, _1, _2, _3, _4, _5, _6, _7, _8))),
- next(next_)
-{
- self=Id(cpg->getLocalNoideId(), getpid());
+Cluster::Cluster(const std::string& name_, const std::string& url_) :
+ name(name_),
+ url(url_),
+ cpg(new Cpg(
+ boost::bind(&Cluster::cpgDeliver, this, _1, _2, _3, _4, _5, _6),
+ boost::bind(&Cluster::cpgConfigChange, this, _1, _2, _3, _4, _5, _6, _7, _8))),
+ self(cpg->getLocalNoideId(), getpid())
+{}
+
+void Cluster::join(FrameHandler::Chain next) {
QPID_LOG(trace, *this << " Joining cluster.");
+ next = next;
+ dispatcher=Thread(*this);
cpg->join(name);
notify();
- dispatcher=Thread(*this);
}
Cluster::~Cluster() {
- try {
- QPID_LOG(trace, *this << " Leaving cluster.");
- cpg->leave(name);
- cpg.reset();
- dispatcher.join();
- } catch (const std::exception& e) {
- QPID_LOG(error, "Exception leaving cluster " << e.what());
+ if (cpg) {
+ try {
+ QPID_LOG(trace, *this << " Leaving cluster.");
+ cpg->leave(name);
+ cpg.reset();
+ dispatcher.join();
+ } catch (const std::exception& e) {
+ QPID_LOG(error, "Exception leaving cluster " << e.what());
+ }
}
}
void Cluster::handle(AMQFrame& frame) {
+ assert(cpg);
QPID_LOG(trace, *this << " SEND: " << frame);
Buffer buf(frame.size());
frame.encode(buf);
@@ -104,52 +126,59 @@
frame.decode(buf);
QPID_LOG(trace, *this << " RECV: " << frame);
// TODO aconway 2007-06-20: use visitor pattern.
- ClusterNotifyBody* notifyIn= dynamic_cast<ClusterNotifyBody*>(frame.getBody().get());
+ ClusterNotifyBody* notifyIn=
+ dynamic_cast<ClusterNotifyBody*>(frame.getBody().get());
if (notifyIn) {
- Mutex::ScopedLock l(lock);
- members[from].reset(new Member(notifyIn->getUrl()));
- lock.notifyAll();
+ {
+ Mutex::ScopedLock l(lock);
+ assert(members[from]);
+ members[from]->url = notifyIn->getUrl();
+ members[from]->status = Member::BROKER;
+ }
+ if (callback)
+ callback();
}
else
- next.handle(frame);
+ next->handle(frame);
}
void Cluster::cpgConfigChange(
cpg_handle_t /*handle*/,
struct cpg_name */*group*/,
- struct cpg_address *ccMembers, int nMembers,
+ struct cpg_address *current, int nCurrent,
struct cpg_address *left, int nLeft,
struct cpg_address *joined, int nJoined
)
{
- QPID_LOG(
- trace,
- *this << " Configuration change. " << endl
- << " Joined: " << make_pair(joined, nJoined) << endl
- << " Left: " << make_pair(left, nLeft) << endl
- << " Current: " << make_pair(ccMembers, nMembers));
+ QPID_LOG(trace,
+ *this << " Configuration change. " << endl
+ << " Joined: " << make_pair(joined, nJoined) << endl
+ << " Left: " << make_pair(left, nLeft) << endl
+ << " Current: " << make_pair(current, nCurrent));
+ bool needNotify=false;
+ MemberList updated;
{
Mutex::ScopedLock l(lock);
- // Erase members that left.
- for (int i = 0; i < nLeft; ++i)
- members.erase(Id(left[i]));
- lock.notifyAll();
- }
-
- // If there are new members (other than myself) then notify.
- for (int i=0; i< nJoined; ++i) {
- if (Id(joined[i]) != self) {
- notify();
- break;
+ for (int i = 0; i < nJoined; ++i) {
+ Id id(current[i]);
+ members[id].reset(new Member(current[i]));
+ if (id != self)
+ needNotify = true; // Notify new members other than myself.
}
- }
-
- // Note: New members are be added to my map when cpgDeliver
- // gets a cluster.notify frame.
+ for (int i = 0; i < nLeft; ++i)
+ members.erase(Id(current[i]));
+ } // End of locked scope.
+ if (needNotify)
+ notify();
+ if (callback)
+ callback();
}
+void Cluster::setCallback(boost::function<void()> f) { callback=f; }
+
void Cluster::run() {
+ assert(cpg);
cpg->dispatchBlocking();
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h Fri Jun 29 10:59:00 2007
@@ -25,47 +25,74 @@
#include "qpid/sys/Monitor.h"
#include "qpid/sys/Runnable.h"
#include "qpid/shared_ptr.h"
-#include "qpid/framing/ProtocolVersion.h"
+#include <boost/function.hpp>
#include <boost/scoped_ptr.hpp>
#include <map>
#include <vector>
namespace qpid {
+
+namespace broker {
+class HandlerUpdater;
+}
+
namespace cluster {
+class ChannelManager;
+
/**
- * Represents a cluster. Creating an instance joins current process
- * to the cluster.
+ * Represents a cluster, provides access to data about members.
+ *
+ * Implements a FrameHandler that multicasts frames to the cluster.
+ *
+ * Requires a handler for frames arriving from the cluster,
+ * normally a ChannelManager but other handlers could be interposed
+ * for testing, logging etc.
*/
class Cluster : public framing::FrameHandler, private sys::Runnable {
public:
/** Details of a cluster member */
struct Member {
- Member(const std::string& url_) : url(url_) {}
+ typedef shared_ptr<const Member> Ptr;
+ /** Status of a cluster member. */
+ enum Status {
+ JOIN, ///< Process joined the group.
+ LEAVE, ///< Process left the group cleanly.
+ NODEDOWN, ///< Process's node went down.
+ NODEUP, ///< Process's node joined the cluster.
+ PROCDOWN, ///< Process died without leaving.
+ BROKER ///< Broker details are available.
+ };
+
+ Member(const cpg_address&);
std::string url;
+ Status status;
};
-
- typedef std::vector<shared_ptr<const Member> > MemberList;
+
+ typedef std::vector<Member::Ptr> MemberList;
/**
- * Join a cluster.
+ * Create a cluster object but do not joing.
* @param name of the cluster.
* @param url of this broker, sent to the cluster.
- * @param next handler receives the frame when it has been
- * acknowledged by the cluster.
*/
- Cluster(const std::string& name,
- const std::string& url,
- framing::FrameHandler& next,
- framing::ProtocolVersion);
+ Cluster(const std::string& name, const std::string& url);
~Cluster();
+
+ /** Join the cluster.
+ *@handler is the handler for frames arriving from the cluster.
+ */
+ void join(framing::FrameHandler::Chain handler);
/** Multicast a frame to the cluster. */
void handle(framing::AMQFrame&);
/** Get the current cluster membership. */
MemberList getMembers() const;
+
+ /** Called when membership changes. */
+ void setCallback(boost::function<void()>);
/** Number of members in the cluster. */
size_t size() const;
@@ -76,7 +103,6 @@
void run();
void notify();
-
void cpgDeliver(
cpg_handle_t /*handle*/,
struct cpg_name *group,
@@ -93,18 +119,14 @@
struct cpg_address */*joined*/, int /*nJoined*/
);
- Id self;
+ mutable sys::Monitor lock;
Cpg::Name name;
std::string url;
- framing::ProtocolVersion version;
boost::scoped_ptr<Cpg> cpg;
- framing::FrameHandler& next;
+ Id self;
MemberMap members;
sys::Thread dispatcher;
-
- protected:
- // Allow access from ClusterTest subclass.
- mutable sys::Monitor lock;
+ boost::function<void()> callback;
friend std::ostream& operator <<(std::ostream&, const Cluster&);
};
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPluginProvider.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPluginProvider.cpp?view=auto&rev=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPluginProvider.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPluginProvider.cpp Fri Jun 29 10:59:00 2007
@@ -0,0 +1,66 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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/broker/Broker.h"
+#include "qpid/framing/HandlerUpdater.h"
+#include "qpid/cluster/Cluster.h"
+#include "qpid/cluster/ChannelManager.h"
+#include "qpid/Plugin.h"
+#include "qpid/Options.h"
+
+namespace qpid {
+namespace cluster {
+
+using namespace std;
+
+struct ClusterPluginProvider : public PluginProvider {
+
+ struct ClusterOptions : public Options {
+ string clusterName;
+ ClusterOptions() {
+ addOptions()
+ ("cluster", optValue(clusterName, "NAME"),
+ "Join the cluster named NAME");
+ }
+ };
+
+ ClusterOptions options;
+ shared_ptr<Cluster> cluster;
+
+ Options* getOptions() {
+ return &options;
+ }
+
+ void provide(PluginUser& user) {
+ broker::Broker* broker = dynamic_cast<broker::Broker*>(&user);
+ // Only provide to a Broker, and only if the --cluster config is set.
+ if (broker && !options.clusterName.empty()) {
+ assert(!cluster); // A process can only belong to one cluster.
+ cluster.reset(new Cluster(options.clusterName, broker->getUrl()));
+
+ // Channel manager is both the next handler for the cluster
+ // and the HandlerUpdater plugin for the broker.
+ shared_ptr<ChannelManager> manager(new ChannelManager(cluster));
+ cluster->join(manager);
+ broker->use(manager);
+ }
+ }
+};
+
+static ClusterPluginProvider instance; // Static initialization.
+
+}} // namespace qpid::cluster
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPluginProvider.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPluginProvider.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cpg.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cpg.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cpg.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cpg.h Fri Jun 29 10:59:00 2007
@@ -23,11 +23,10 @@
#include "qpid/cluster/Dispatchable.h"
#include <boost/function.hpp>
#include <cassert>
-#ifdef CLUSTER
extern "C" {
#include <openais/cpg.h>
}
-#endif
+
namespace qpid {
namespace cluster {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp Fri Jun 29 10:59:00 2007
@@ -39,19 +39,13 @@
assert(version != ProtocolVersion(0,0));
}
-AMQFrame::AMQFrame(ProtocolVersion _version, uint16_t _channel, AMQBody* _body) :
-version(_version), channel(_channel), body(_body)
-{}
+AMQFrame::AMQFrame(ProtocolVersion _version, uint16_t _channel, AMQBody* _body) : channel(_channel), body(_body),version(_version) {}
AMQFrame::AMQFrame(ProtocolVersion _version, uint16_t _channel, const AMQBody::shared_ptr& _body) :
-version(_version), channel(_channel), body(_body)
+ channel(_channel), body(_body), version(_version)
{}
AMQFrame::~AMQFrame() {}
-
-uint16_t AMQFrame::getChannel(){
- return channel;
-}
AMQBody::shared_ptr AMQFrame::getBody(){
return body;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h Fri Jun 29 10:59:00 2007
@@ -48,7 +48,7 @@
virtual void encode(Buffer& buffer);
virtual bool decode(Buffer& buffer);
virtual uint32_t size() const;
- uint16_t getChannel();
+ uint16_t getChannel() const { return channel; }
AMQBody::shared_ptr getBody();
/** Convenience template to cast the body to an expected type */
@@ -60,17 +60,16 @@
uint32_t decodeHead(Buffer& buffer);
void decodeBody(Buffer& buffer, uint32_t size);
- private:
- static AMQP_MethodVersionMap versionMap;
- ProtocolVersion version;
-
uint16_t channel;
uint8_t type;
AMQBody::shared_ptr body;
-
+ ProtocolVersion version;
- friend std::ostream& operator<<(std::ostream& out, const AMQFrame& body);
+ private:
+ static AMQP_MethodVersionMap versionMap;
};
+
+std::ostream& operator<<(std::ostream&, const AMQFrame&);
}} // namespace qpid::framing
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h Fri Jun 29 10:59:00 2007
@@ -40,11 +40,21 @@
Chain out;
};
+ Handler() {}
+ Handler(Chain next_) : next(next_) {}
virtual ~Handler() {}
+
virtual void handle(T) = 0;
+
+ /** Next handler. Public so chains can be modified by altering next. */
Chain next;
-};
+ protected:
+ /** Derived handle() implementations call nextHandler to invoke the
+ * next handler in the chain. */
+ void nextHandler(T data) { if (next) next->handle(data); }
+
+};
}}
Copied: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/HandlerUpdater.h (from r551363, incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ChannelInitializer.h)
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/HandlerUpdater.h?view=diff&rev=551981&p1=incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ChannelInitializer.h&r1=551363&p2=incubator/qpid/trunk/qpid/cpp/src/qpid/framing/HandlerUpdater.h&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ChannelInitializer.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/HandlerUpdater.h Fri Jun 29 10:59:00 2007
@@ -1,5 +1,5 @@
-#ifndef QPID_BROKER_CHANNELINITIALIZER_H
-#define QPID_BROKER_CHANNELINITIALIZER_H
+#ifndef QPID_FRAMING_HANDLERUPDATER_H
+#define QPID_FRAMING_HANDLERUPDATER_H
/*
*
@@ -19,25 +19,26 @@
*
*/
-#include <boost/noncopyable.hpp>
+#include "qpid/Plugin.h"
+#include "qpid/framing/amqp_types.h"
+#include "qpid/framing/FrameHandler.h"
namespace qpid {
-namespace broker {
+namespace framing {
-/**
- * A ChannelInitializer is called each time a new Channel is created.
- */
-class ChannelInitializer : boost::noncopyable
-{
- public:
- virtual ~ChannelInitializer() {}
-
- /** Called for each new channel */
- virtual initialize(Channe&) = 0;
+/** Plugin object that can update handler chains. */
+struct HandlerUpdater : public Plugin {
+ /** Update the handler chains.
+ *@param id Unique identifier for channel or session.
+ *@param chains Handler chains to be updated.
+ */
+ virtual void update(ChannelId id, FrameHandler::Chains& chains) = 0;
};
-}} // namespace qpid::broker
+}} // namespace qpid::framing
+
+
-#endif /*!QPID_BROKER_CHANNELINITIALIZER_H*/
+#endif /*!QPID_FRAMING_HANDLERUPDATER_H*/
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h Fri Jun 29 10:59:00 2007
@@ -53,5 +53,12 @@
// Types represented by classes.
class Content;
class FieldTable;
+
+// Useful constants
+
+/** Maximum channel ID used by broker. Reserve high bit for internal use.*/
+const ChannelId CHANNEL_MAX=(ChannelId(~1))>>1;
+const ChannelId CHANNEL_HIGH_BIT= ChannelId(~CHANNEL_MAX);
+
}} // namespace qpid::framing
#endif
Modified: incubator/qpid/trunk/qpid/cpp/src/qpidd.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpidd.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpidd.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpidd.cpp Fri Jun 29 10:59:00 2007
@@ -74,7 +74,7 @@
};
// Globals
-Broker::shared_ptr brokerPtr;
+shared_ptr<Broker> brokerPtr;
QpiddOptions options;
void handle_signal(int /*signal*/){
@@ -112,7 +112,7 @@
/** Code for forked child */
void child(Daemon& demon) {
- brokerPtr=Broker::create(options.broker);
+ brokerPtr.reset(new Broker(options.broker));
uint16_t realPort=brokerPtr->getPort();
demon.ready(realPort); // Notify parent.
brokerPtr->run();
@@ -161,7 +161,7 @@
demon.fork(parent, child);
}
else { // Non-daemon broker.
- brokerPtr = Broker::create(options.broker);
+ brokerPtr.reset(new Broker(options.broker));
if (options.broker.port == 0)
cout << uint16_t(brokerPtr->getPort()) << endl;
brokerPtr->run();
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/.valgrindrc-default
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/.valgrindrc-default?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/.valgrindrc-default (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/.valgrindrc-default Fri Jun 29 10:59:00 2007
@@ -3,5 +3,5 @@
--demangle=yes
--suppressions=.valgrind.supp
--num-callers=25
---track-fds=yes
+--trace-children=yes
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/BrokerChannelTest.cpp Fri Jun 29 10:59:00 2007
@@ -60,7 +60,7 @@
CPPUNIT_TEST(testFlow);
CPPUNIT_TEST_SUITE_END();
- Broker::shared_ptr broker;
+ shared_ptr<Broker> broker;
Connection connection;
MockHandler handler;
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.cpp Fri Jun 29 10:59:00 2007
@@ -24,46 +24,44 @@
#include "qpid/framing/BasicGetOkBody.h"
-
static const ProtocolVersion VER;
/** Verify membership ind a cluster with one member. */
BOOST_AUTO_TEST_CASE(clusterOne) {
- VectorFrameHandler received;
- Cluster cluster("Test", "amqp:one:1", received, VER);
+ Cluster cluster("Test", "amqp:one:1");
+ TestClusterHandler handler(cluster);
AMQFrame frame(VER, 1, new ChannelOkBody(VER));
-
cluster.handle(frame);
- BOOST_REQUIRE(received.waitFor(1));
+ BOOST_REQUIRE(handler.waitFrames(1));
BOOST_CHECK_EQUAL(1u, cluster.size());
Cluster::MemberList members = cluster.getMembers();
BOOST_CHECK_EQUAL(1u, members.size());
BOOST_REQUIRE_EQUAL(members.front()->url, "amqp:one:1");
- BOOST_CHECK_EQUAL(1u, received.size());
- BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *received[0].getBody());
+ BOOST_CHECK_EQUAL(1u, handler.size());
+ BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *handler[0].getBody());
}
/** Fork a process to verify membership in a cluster with two members */
BOOST_AUTO_TEST_CASE(clusterTwo) {
- VectorFrameHandler received;
pid_t pid=fork();
BOOST_REQUIRE(pid >= 0);
- if (pid) { // Parent
- TestCluster cluster("Test", "amqp::1", received, VER);
- BOOST_REQUIRE(cluster.waitFor(2));
+ if (pid) { // Parent see Cluster_child.cpp for child.
+ Cluster cluster("Test", "amqp::1");
+ TestClusterHandler handler(cluster);
+ BOOST_REQUIRE(handler.waitMembers(2));
// Exchange frames with child.
AMQFrame frame(VER, 1, new ChannelOkBody(VER));
cluster.handle(frame);
- BOOST_REQUIRE(received.waitFor(2));
- BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *received[0].getBody());
- BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *received[1].getBody());
+ BOOST_REQUIRE(handler.waitFrames(2));
+ BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *handler[0].getBody());
+ BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *handler[1].getBody());
// Wait for child to exit.
int status;
BOOST_CHECK_EQUAL(::wait(&status), pid);
BOOST_CHECK_EQUAL(0, status);
- BOOST_CHECK(cluster.waitFor(1));
+ BOOST_CHECK(handler.waitMembers(1));
BOOST_CHECK_EQUAL(1u, cluster.size());
}
else { // Child
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Cluster.h Fri Jun 29 10:59:00 2007
@@ -24,6 +24,7 @@
#include "qpid/framing/ChannelOkBody.h"
#include "qpid/framing/BasicGetOkBody.h"
#include "qpid/log/Logger.h"
+#include <boost/bind.hpp>
#include <iostream>
#include <vector>
@@ -39,44 +40,44 @@
using namespace qpid::framing;
using namespace qpid::sys;
-struct TestCluster : public Cluster {
- TestCluster(const std::string& name,
- const std::string& url,
- framing::FrameHandler& next,
- framing::ProtocolVersion ver) : Cluster(name,url,next, ver) {}
-
- /** Wait for the cluster to be of expected size (exactly) */
- bool waitFor(size_t n) {
- Mutex::ScopedLock l(lock);
- AbsTime deadline(now(),2*TIME_SEC);
- while(size() != n && lock.wait(deadline))
- ;
- return size() == n;
- }
-};
-
-struct VectorFrameHandler :
+void null_deleter(void*) {}
+
+struct TestClusterHandler :
public std::vector<AMQFrame>, public FrameHandler, public Monitor
{
+ TestClusterHandler(Cluster& c) : cluster(c) {
+ cluster.join(make_shared_ptr(this, &null_deleter));
+ cluster.setCallback(boost::bind(&Monitor::notify, this));
+ }
+
void handle(AMQFrame& f) {
ScopedLock l(*this);
push_back(f);
notifyAll();
}
- /** Wait for vector to reach size n exactly */
- bool waitFor(size_t n) {
+ /** Wait for the vector to contain n frames. */
+ bool waitFrames(size_t n) {
ScopedLock l(*this);
- AbsTime deadline(now(), 1*TIME_SEC);
+ AbsTime deadline(now(), TIME_SEC);
while (size() != n && wait(deadline))
;
return size() == n;
}
-};
+ /** Wait for the cluster to have n members */
+ bool waitMembers(size_t n) {
+ ScopedLock l(*this);
+ AbsTime deadline(now(), TIME_SEC);
+ while (cluster.size() != n && wait(deadline))
+ ;
+ return cluster.size() == n;
+ }
+
+ Cluster& cluster;
+};
-// namespace
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp Fri Jun 29 10:59:00 2007
@@ -32,16 +32,14 @@
/** Chlid part of Cluster::clusterTwo test */
void clusterTwo() {
- VectorFrameHandler received;
- TestCluster cluster("Test", "amqp::2", received, VER);
- BOOST_REQUIRE(cluster.waitFor(2));
-
- BOOST_REQUIRE(received.waitFor(1));
- BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *received[0].getBody());
+ Cluster cluster("Test", "amqp::2");
+ TestClusterHandler handler(cluster);
+ BOOST_REQUIRE(handler.waitFrames(1));
+ BOOST_CHECK_TYPEID_EQUAL(ChannelOkBody, *handler[0].getBody());
AMQFrame frame(VER, 1, new BasicGetOkBody(VER));
cluster.handle(frame);
- BOOST_REQUIRE(received.waitFor(2));
- BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *received[1].getBody());
+ BOOST_REQUIRE(handler.waitFrames(2));
+ BOOST_CHECK_TYPEID_EQUAL(BasicGetOkBody, *handler[1].getBody());
}
int test_main(int, char**) {
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp Fri Jun 29 10:59:00 2007
@@ -402,8 +402,8 @@
broker::InProcessBroker::Conversation::const_iterator i = ibroker.conversation.begin();
ASSERT_FRAME("BROKER: Frame[channel=0; request(id=1,mark=0): ConnectionStart: versionMajor=0; versionMinor=9; serverProperties={}; mechanisms=PLAIN; locales=en_US]", *i++);
ASSERT_FRAME("CLIENT: Frame[channel=0; response(id=1,request=1,batch=0): ConnectionStartOk: clientProperties={}; mechanism=PLAIN; response=\000guest\000guest; locale=en_US]", *i++);
- ASSERT_FRAME("BROKER: Frame[channel=0; request(id=2,mark=1): ConnectionTune: channelMax=100; frameMax=65536; heartbeat=0]", *i++);
- ASSERT_FRAME("CLIENT: Frame[channel=0; response(id=2,request=2,batch=0): ConnectionTuneOk: channelMax=100; frameMax=65536; heartbeat=0]", *i++);
+ ASSERT_FRAME("BROKER: Frame[channel=0; request(id=2,mark=1): ConnectionTune: channelMax=32767; frameMax=65536; heartbeat=0]", *i++);
+ ASSERT_FRAME("CLIENT: Frame[channel=0; response(id=2,request=2,batch=0): ConnectionTuneOk: channelMax=32767; frameMax=65536; heartbeat=0]", *i++);
ASSERT_FRAME("CLIENT: Frame[channel=0; request(id=1,mark=0): ConnectionOpen: virtualHost=/; capabilities=; insist=1]", *i++);
ASSERT_FRAME("BROKER: Frame[channel=0; response(id=1,request=1,batch=0): ConnectionOpenOk: knownHosts=]", *i++);
ASSERT_FRAME("CLIENT: Frame[channel=1; request(id=1,mark=0): ChannelOpen: outOfBand=]", *i++);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/InProcessBroker.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/InProcessBroker.h?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/InProcessBroker.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/InProcessBroker.h Fri Jun 29 10:59:00 2007
@@ -113,7 +113,7 @@
};
framing::ProtocolInitiation protocolInit;
- Broker::shared_ptr broker;
+ shared_ptr<Broker> broker;
OutputToInputHandler brokerOut;
broker::Connection brokerConnection;
OutputToInputHandler clientOut;
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am Fri Jun 29 10:59:00 2007
@@ -83,7 +83,7 @@
topic_publisher
-check_PROGRAMS += $(unit_progs) $(testprogs) interop_runner
+check_PROGRAMS += $(unit_progs) $(testprogs) interop_runner
TESTS_ENVIRONMENT = VALGRIND=$(VALGRIND) srcdir=$(srcdir) $(srcdir)/run_test
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Url.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Url.cpp?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Url.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Url.cpp Fri Jun 29 10:59:00 2007
@@ -30,27 +30,33 @@
Url url;
url.push_back(TcpAddress("foo.com"));
url.push_back(TcpAddress("bar.com", 6789));
-
- BOOST_CHECK_EQUAL(
- url.str(), "amqp:tcp:foo.com:5672,tcp:bar.com:6789");
- BOOST_CHECK_EQUAL(Url().str(), "amqp:");
+ BOOST_CHECK_EQUAL("amqp:tcp:foo.com:5672,tcp:bar.com:6789", url.str());
+ BOOST_CHECK_EQUAL("amqp:", Url().str());
}
-BOOST_AUTO_TEST_CASE(testUrl_ctor) {
- BOOST_CHECK_EQUAL(
- Url("amqp:foo.com,tcp:bar.com:1234").str(),
- "amqp:tcp:foo.com:5672,tcp:bar.com:1234");
- BOOST_CHECK_EQUAL(
- Url("amqp:foo/ignorethis").str(),
- "amqp:tcp:foo:5672");
- BOOST_CHECK_EQUAL("amqp:tcp::5672", Url("amqp:").str());
- BOOST_CHECK_EQUAL(0u, Url("xxx", nothrow).size());
+BOOST_AUTO_TEST_CASE(testUrl_parse) {
+ Url url;
+ url.parse("amqp:foo.com,tcp:bar.com:1234");
+ BOOST_CHECK_EQUAL(2u, url.size());
+ BOOST_CHECK_EQUAL("foo.com", boost::get<TcpAddress>(url[0]).host);
+ BOOST_CHECK_EQUAL("amqp:tcp:foo.com:5672,tcp:bar.com:1234", url.str());
+
+ url.parse("amqp:foo/ignorethis");
+ BOOST_CHECK_EQUAL("amqp:tcp:foo:5672", url.str());
+
+ url.parse("amqp:");
+ BOOST_CHECK_EQUAL("amqp:tcp::5672", url.str());
+
try {
- Url invalid("xxx");
+ url.parse("invalid url");
BOOST_FAIL("Expected InvalidUrl exception");
}
catch (const Url::InvalidUrl&) {}
+
+ url.parseNoThrow("invalid url");
+ BOOST_CHECK(url.empty());
}
+
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/run_test
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/run_test?view=diff&rev=551981&r1=551980&r2=551981
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/run_test (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/run_test Fri Jun 29 10:59:00 2007
@@ -47,7 +47,7 @@
# This is a libtool "executable". Valgrind it if VALGRIND specified.
test -n "$VALGRIND" && VALGRIND="$VALGRIND --log-file-exactly=$VG_LOG --"
# Hide output unless there's an error.
- libtool --mode=execute "$VALGRIND" "$@" >$TEST_LOG 2>&1 || {
+ libtool --mode=execute $VALGRIND "$@" >$TEST_LOG 2>&1 || {
ERROR=$?
cat $TEST_LOG
}