diff options
Diffstat (limited to 'cpp/src/qpid')
| -rw-r--r-- | cpp/src/qpid/CommonOptions.cpp | 36 | ||||
| -rw-r--r-- | cpp/src/qpid/CommonOptions.h | 91 | ||||
| -rw-r--r-- | cpp/src/qpid/Exception.cpp | 6 | ||||
| -rw-r--r-- | cpp/src/qpid/Exception.h | 4 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Broker.cpp | 55 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Broker.h | 28 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Configuration.cpp | 252 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/Configuration.h | 171 |
8 files changed, 194 insertions, 449 deletions
diff --git a/cpp/src/qpid/CommonOptions.cpp b/cpp/src/qpid/CommonOptions.cpp new file mode 100644 index 0000000000..8efb3a6baa --- /dev/null +++ b/cpp/src/qpid/CommonOptions.cpp @@ -0,0 +1,36 @@ +/* + * + * 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 "CommonOptions.h" + +namespace qpid { + +const int CommonOptions::DEFAULT_PORT=5672; + +CommonOptions::CommonOptions() : trace(false), port(DEFAULT_PORT) {} + +void CommonOptions::addTo(po::options_description& desc) +{ + using namespace po; + desc.add_options() + ("trace,t", optValue(trace), "Enable debug tracing" ) + ("port,p", optValue(port,"PORT"), "Use PORT for AMQP connections."); +} + +} // namespace qpid + diff --git a/cpp/src/qpid/CommonOptions.h b/cpp/src/qpid/CommonOptions.h new file mode 100644 index 0000000000..2a4b9798cd --- /dev/null +++ b/cpp/src/qpid/CommonOptions.h @@ -0,0 +1,91 @@ +#ifndef QPID_COMMONOPTIONS_H +#define QPID_COMMONOPTIONS_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/program_options.hpp> +#include <boost/format.hpp> + +namespace qpid { + +/**@Qpid extensions to boost::program_options */ +namespace program_options { + +using namespace boost::program_options; + +/** @internal Normally only constructed by optValue() */ +template <class T> +class OptionValue : public typed_value<T> { + public: + OptionValue(T& value, const std::string& arg) + : typed_value<T>(&value), argName(arg) {} + std::string name() const { return argName; } + private: + std::string argName; +}; + +/** + * Like boost::program_options::value() with more convenient signature + * for updating a value by reference and nicer help formatting. + * + *@param value displayed as default in help, updated from options. + * Must support ostream << operator. + *@param arg name for arguments in help. + * + *@see CommonOptions.cpp for example of use. + */ +template<class T> +value_semantic* optValue(T& value, const char* arg) { + std::string val(boost::lexical_cast<std::string>(value)); + std::string argName( + val.empty() ? std::string(arg) : + (boost::format("%s (=%s) ") % arg % val).str()); + return new OptionValue<T>(value, argName); +} + +/** + * Like boost::program_options::bool_switch but takes reference, not pointer. + */ +inline value_semantic* optValue(bool& value) { return bool_switch(&value); } + +} // namespace program_options + +namespace po=program_options; // Convenience alias. + +/** + * Common options for client and broker + */ +struct CommonOptions { + static const int DEFAULT_PORT; + + CommonOptions(); + + bool trace; + int port; + + /** Add members to program_options to be updated */ + void addTo(po::options_description&); +}; + +} // namespace qpid + +#endif /*!QPID_COMMONOPTIONS_H*/ diff --git a/cpp/src/qpid/Exception.cpp b/cpp/src/qpid/Exception.cpp index 1bbf89e99f..e73f64f2ba 100644 --- a/cpp/src/qpid/Exception.cpp +++ b/cpp/src/qpid/Exception.cpp @@ -20,9 +20,15 @@ */ #include "Exception.h" +#include <errno.h> namespace qpid { +std::string strError(int err) { + char buf[512]; + return std::string(strerror_r(err, buf, sizeof(buf))); +} + Exception::Exception() throw() {} Exception::Exception(const std::string& str) throw() : whatStr(str) {} diff --git a/cpp/src/qpid/Exception.h b/cpp/src/qpid/Exception.h index 4fbad7efaa..13583042a8 100644 --- a/cpp/src/qpid/Exception.h +++ b/cpp/src/qpid/Exception.h @@ -31,6 +31,10 @@ namespace qpid { + +/** Get the error message for error number err. */ +std::string strError(int err); + /** * Exception base class for all Qpid exceptions. */ diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp index cda3745522..b285fbad24 100644 --- a/cpp/src/qpid/broker/Broker.cpp +++ b/cpp/src/qpid/broker/Broker.cpp @@ -40,13 +40,38 @@ namespace qpid { namespace broker { +Broker::Options::Options() : + workerThreads(5), + maxConnections(500), + connectionBacklog(10), + store(), + stagingThreshold(5000000) +{} + +void Broker::Options::addTo(po::options_description& desc) +{ + using namespace po; + CommonOptions::addTo(desc); + desc.add_options() + ("worker-threads", optValue(workerThreads, "N"), + "Broker thread pool size") + ("max-connections", optValue(maxConnections, "N"), + "Maximum allowed connections") + ("connection-backlog", optValue(connectionBacklog, "N"), + "Connection backlog limit for server socket.") + ("staging-threshold", optValue(stagingThreshold, "N"), + "Messages over N bytes are staged to disk.") + ("store", optValue(store,"LIBNAME"), + "Name of message store shared library."); +} + const std::string empty; const std::string amq_direct("amq.direct"); const std::string amq_topic("amq.topic"); const std::string amq_fanout("amq.fanout"); const std::string amq_match("amq.match"); -Broker::Broker(const Configuration& conf) : +Broker::Broker(const Broker::Options& conf) : config(conf), store(createStore(conf)), queues(store.get()), @@ -63,7 +88,8 @@ Broker::Broker(const Configuration& conf) : exchanges.declare(amq_match, HeadersExchange::typeName); if(store.get()) { - RecoveryManagerImpl recoverer(queues, exchanges, conf.getStagingThreshold()); + RecoveryManagerImpl recoverer( + queues, exchanges, conf.stagingThreshold); store->recover(recoverer); } @@ -73,20 +99,20 @@ Broker::Broker(const Configuration& conf) : Broker::shared_ptr Broker::create(int16_t port) { - Configuration config; - config.setPort(port); + Options config; + config.port=port; return create(config); } -Broker::shared_ptr Broker::create(const Configuration& config) { +Broker::shared_ptr Broker::create(const Options& config) { return Broker::shared_ptr(new Broker(config)); } -MessageStore* Broker::createStore(const Configuration& config) { - if (config.getStore().empty()) - return new NullMessageStore(config.isTrace()); +MessageStore* Broker::createStore(const Options& config) { + if (config.store.empty()) + return new NullMessageStore(config.trace); else - return new MessageStoreModule(config.getStore()); + return new MessageStoreModule(config.store); } void Broker::run() { @@ -108,16 +134,13 @@ int16_t Broker::getPort() const { return getAcceptor().getPort(); } Acceptor& Broker::getAcceptor() const { if (!acceptor) const_cast<Acceptor::shared_ptr&>(acceptor) = - Acceptor::create(config.getPort(), - config.getConnectionBacklog(), - config.getWorkerThreads(), - config.isTrace()); + Acceptor::create(config.port, + config.connectionBacklog, + config.workerThreads, + config.trace); return *acceptor; } -const int16_t Broker::DEFAULT_PORT(5672); - - }} // namespace qpid::broker diff --git a/cpp/src/qpid/broker/Broker.h b/cpp/src/qpid/broker/Broker.h index e732664acb..187167ebe7 100644 --- a/cpp/src/qpid/broker/Broker.h +++ b/cpp/src/qpid/broker/Broker.h @@ -22,7 +22,6 @@ * */ -#include "Configuration.h" #include "ConnectionFactory.h" #include "qpid/sys/Runnable.h" #include "qpid/sys/Acceptor.h" @@ -36,8 +35,9 @@ #include "qpid/framing/OutputHandler.h" #include "qpid/framing/ProtocolInitiation.h" #include "QueueRegistry.h" +#include "qpid/CommonOptions.h" -namespace qpid { +namespace qpid { namespace broker { /** * A broker instance. @@ -46,20 +46,28 @@ class Broker : public sys::Runnable, public SharedObject<Broker> { public: - static const int16_t DEFAULT_PORT; - + struct Options : public CommonOptions { + Options(); + void addTo(po::options_description&); + int workerThreads; + int maxConnections; + int connectionBacklog; + std::string store; + long stagingThreshold; + }; + 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 = DEFAULT_PORT); + static shared_ptr create(int16_t port = CommonOptions::DEFAULT_PORT); /** - * Create a broker using a Configuration. + * Create a broker with the options in config. */ - static shared_ptr create(const Configuration& config); + static shared_ptr create(const Options& config); /** * Return listening port. If called before bind this is @@ -87,10 +95,10 @@ class Broker : public sys::Runnable, DtxManager& getDtxManager() { return dtxManager; } private: - Broker(const Configuration& config); + Broker(const Options& configuration); sys::Acceptor& getAcceptor() const; - Configuration config; + Options config; sys::Acceptor::shared_ptr acceptor; const std::auto_ptr<MessageStore> store; QueueRegistry queues; @@ -101,7 +109,7 @@ class Broker : public sys::Runnable, ConnectionFactory factory; DtxManager dtxManager; - static MessageStore* createStore(const Configuration& config); + static MessageStore* createStore(const Options& config); }; }} diff --git a/cpp/src/qpid/broker/Configuration.cpp b/cpp/src/qpid/broker/Configuration.cpp deleted file mode 100644 index 3449f6f179..0000000000 --- a/cpp/src/qpid/broker/Configuration.cpp +++ /dev/null @@ -1,252 +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 "Configuration.h" -#include <string.h> -#include "config.h" - -using namespace qpid::broker; -using namespace std; - -Configuration::Configuration() : - daemon('d', "daemon", "Run as system daemon, detached from terminal.", false), - trace('t', "trace", "Print incoming & outgoing frames to the console", false), - port('p', "port", "Set the port to listen on (default=5672)", 5672), - workerThreads("worker-threads", "Set the number of worker threads to use (default=5).", 5), - maxConnections("max-connections", "Set the maximum number of connections the broker can accept (default=500).", 500), - connectionBacklog("connection-backlog", "Set the connection backlog for the servers socket (default=10)", 10), - store('s', "store", "Set the message store module to use (default='' which implies no store)", ""), - stagingThreshold("staging-threshold", "Set the message size threshold above which messages will be written to disk as they arrive (default=5,000,000)", 5000000), - help("help", "Print usage information", false), - version("version", "Print version information", false) -{ - options.push_back(&daemon); - options.push_back(&trace); - options.push_back(&port); - options.push_back(&workerThreads); - options.push_back(&maxConnections); - options.push_back(&connectionBacklog); - options.push_back(&store); - options.push_back(&stagingThreshold); - options.push_back(&help); - options.push_back(&version); -} - -Configuration::~Configuration(){} - -void Configuration::parse(char const *progName, int argc, char** argv){ - programName = progName; - int position = 1; - while(position < argc){ - bool matched(false); - for(op_iterator i = options.begin(); i < options.end() && !matched; i++){ - matched = (*i)->parse(position, argv, argc); - } - if(!matched) { - throw BadOptionException( - std::string("Unrecognised option: ")+argv[position]); - } - } -} - -void Configuration::usage(){ - std::cout << "Usage: " << programName << " [OPTION]..." << std::endl - << "Start the Qpid AMQP broker daemon." << std::endl << std::endl - << "Options:" << std::endl; - for(op_iterator i = options.begin(); i < options.end(); i++){ - (*i)->print(std::cout); - } - - std::cout << std::endl << "Report bugs to <" << PACKAGE_BUGREPORT << ">." - << std::endl; -} - -bool Configuration::isHelp() const { - return help.getValue(); -} - -bool Configuration::isVersion() const { - return version.getValue(); -} - -bool Configuration::isDaemon() const { - return daemon.getValue(); -} - -bool Configuration::isTrace() const { - return trace.getValue(); -} - -int Configuration::getPort() const { - return port.getValue(); -} - -int Configuration::getWorkerThreads() const { - return workerThreads.getValue(); -} - -int Configuration::getMaxConnections() const { - return maxConnections.getValue(); -} - -int Configuration::getConnectionBacklog() const { - return connectionBacklog.getValue(); -} - -const std::string& Configuration::getStore() const { - return store.getValue(); -} - -long Configuration::getStagingThreshold() const { - return stagingThreshold.getValue(); -} - - -Configuration::Option::Option(const char _flag, const string& _name, const string& _desc) : - flag(string("-") + _flag), name("--" +_name), desc(_desc) {} - -Configuration::Option::Option(const string& _name, const string& _desc) : - flag(""), name("--" + _name), desc(_desc) {} - -Configuration::Option::~Option(){} - -bool Configuration::Option::match(const string& arg){ - return flag == arg || name == arg; -} - -bool Configuration::Option::parse(int& i, char** argv, int argc){ - const string arg(argv[i]); - if(match(arg)){ - if(needsValue()){ - if(++i < argc) setValue(argv[i]); - else throw ParseException("Argument " + arg + " requires a value!"); - }else{ - setValue(""); - } - i++; - return true; - }else{ - return false; - } -} - -void Configuration::Option::print(ostream& out) const { - out << " "; - if(flag.length() > 0){ - out << flag << ", "; - } else { - out << " "; - } - out << name; - if(needsValue()) out << " <value>"; - out << std::endl; - out << " " << desc << std::endl; -} - - -// String Option: - -Configuration::StringOption::StringOption(const char _flag, const string& _name, const string& _desc, const string _value) : - Option(_flag,_name,_desc), defaultValue(_value), value(_value) {} - -Configuration::StringOption::StringOption(const string& _name, const string& _desc, const string _value) : - Option(_name,_desc), defaultValue(_value), value(_value) {} - -Configuration::StringOption::~StringOption(){} - -const string& Configuration::StringOption::getValue() const { - return value; -} - -bool Configuration::StringOption::needsValue() const { - return true; -} - -void Configuration::StringOption::setValue(const std::string& _value){ - value = _value; -} - -// Int Option: - -Configuration::IntOption::IntOption(const char _flag, const string& _name, const string& _desc, const int _value) : - Option(_flag,_name,_desc), defaultValue(_value), value(_value) {} - -Configuration::IntOption::IntOption(const string& _name, const string& _desc, const int _value) : - Option(_name,_desc), defaultValue(_value), value(_value) {} - -Configuration::IntOption::~IntOption(){} - -int Configuration::IntOption::getValue() const { - return value; -} - -bool Configuration::IntOption::needsValue() const { - return true; -} - -void Configuration::IntOption::setValue(const std::string& _value){ - value = atoi(_value.c_str()); -} - -// Long Option: - -Configuration::LongOption::LongOption(const char _flag, const string& _name, const string& _desc, const long _value) : - Option(_flag,_name,_desc), defaultValue(_value), value(_value) {} - -Configuration::LongOption::LongOption(const string& _name, const string& _desc, const long _value) : - Option(_name,_desc), defaultValue(_value), value(_value) {} - -Configuration::LongOption::~LongOption(){} - -long Configuration::LongOption::getValue() const { - return value; -} - -bool Configuration::LongOption::needsValue() const { - return true; -} - -void Configuration::LongOption::setValue(const std::string& _value){ - value = atol(_value.c_str()); -} - -// Bool Option: - -Configuration::BoolOption::BoolOption(const char _flag, const string& _name, const string& _desc, const bool _value) : - Option(_flag,_name,_desc), defaultValue(_value), value(_value) {} - -Configuration::BoolOption::BoolOption(const string& _name, const string& _desc, const bool _value) : - Option(_name,_desc), defaultValue(_value), value(_value) {} - -Configuration::BoolOption::~BoolOption(){} - -bool Configuration::BoolOption::getValue() const { - return value; -} - -bool Configuration::BoolOption::needsValue() const { - return false; -} - -void Configuration::BoolOption::setValue(const std::string& /*not required*/){ - //BoolOptions have no value. The fact that the option is specified - //implies the value is true. - value = true; -} diff --git a/cpp/src/qpid/broker/Configuration.h b/cpp/src/qpid/broker/Configuration.h deleted file mode 100644 index 174a061637..0000000000 --- a/cpp/src/qpid/broker/Configuration.h +++ /dev/null @@ -1,171 +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. - * - */ -#ifndef _Configuration_ -#define _Configuration_ - -#include <cstdlib> -#include <iostream> -#include <vector> -#include "qpid/Exception.h" - -namespace qpid { -namespace broker { -class Configuration{ - - class Option { - const std::string flag; - const std::string name; - const std::string desc; - - bool match(const std::string& arg); - - protected: - virtual bool needsValue() const = 0; - virtual void setValue(const std::string& value) = 0; - - public: - Option(const char flag, const std::string& name, const std::string& desc); - Option(const std::string& name, const std::string& desc); - virtual ~Option(); - - bool parse(int& i, char** argv, int argc); - void print(std::ostream& out) const; - }; - - class IntOption : public Option{ - const int defaultValue; - int value; - public: - IntOption(char flag, const std::string& name, const std::string& desc, const int value = 0); - IntOption(const std::string& name, const std::string& desc, const int value = 0); - virtual ~IntOption(); - - int getValue() const; - virtual bool needsValue() const; - virtual void setValue(const std::string& value); - virtual void setValue(int _value) { value = _value; } - }; - - class LongOption : public Option{ - const long defaultValue; - int value; - public: - LongOption(char flag, const std::string& name, const std::string& desc, const long value = 0); - LongOption(const std::string& name, const std::string& desc, const long value = 0); - virtual ~LongOption(); - - long getValue() const; - virtual bool needsValue() const; - virtual void setValue(const std::string& value); - virtual void setValue(int _value) { value = _value; } - }; - - class StringOption : public Option{ - const std::string defaultValue; - std::string value; - public: - StringOption(char flag, const std::string& name, const std::string& desc, const std::string value = ""); - StringOption(const std::string& name, const std::string& desc, const std::string value = ""); - virtual ~StringOption(); - - const std::string& getValue() const; - virtual bool needsValue() const; - virtual void setValue(const std::string& value); - }; - - class BoolOption : public Option{ - const bool defaultValue; - bool value; - public: - BoolOption(char flag, const std::string& name, const std::string& desc, const bool value = 0); - BoolOption(const std::string& name, const std::string& desc, const bool value = 0); - virtual ~BoolOption(); - - bool getValue() const; - virtual bool needsValue() const; - virtual void setValue(const std::string& value); - virtual void setValue(bool _value) { value = _value; } - }; - - BoolOption daemon; - BoolOption trace; - IntOption port; - IntOption workerThreads; - IntOption maxConnections; - IntOption connectionBacklog; - StringOption store; - LongOption stagingThreshold; - BoolOption help; - BoolOption version; - char const *programName; - - typedef std::vector<Option*>::iterator op_iterator; - std::vector<Option*> options; - - public: - - struct BadOptionException : public Exception { - template<class T> - BadOptionException(const T& msg) : Exception(msg) {} - }; - - - class ParseException : public Exception { - public: - template <class T> - ParseException(const T& msg) : Exception(msg) {} - }; - - - Configuration(); - ~Configuration(); - - void parse(char const*, int argc, char** argv); - - bool isHelp() const; - bool isVersion() const; - bool isDaemon() const; - bool isTrace() const; - int getPort() const; - int getWorkerThreads() const; - int getMaxConnections() const; - int getConnectionBacklog() const; - const std::string& getStore() const; - long getStagingThreshold() const; - - void setHelp(bool b) { help.setValue(b); } - void setVersion(bool b) { version.setValue(b); } - void setDaemon(bool b) { daemon.setValue(b); } - void setTrace(bool b) { trace.setValue(b); } - void setPort(int i) { port.setValue(i); } - void setWorkerThreads(int i) { workerThreads.setValue(i); } - void setMaxConnections(int i) { maxConnections.setValue(i); } - void setConnectionBacklog(int i) { connectionBacklog.setValue(i); } - void setStore(const std::string& s) { store.setValue(s); } - void setStagingThreshold(long l) { stagingThreshold.setValue(l); } - - void usage(); -}; -} -} - - -#endif |