Author: aconway
Date: Fri Feb 1 08:03:02 2008
New Revision: 617533
URL: http://svn.apache.org/viewvc?rev=617533&view=rev
Log:
Added cluster URL configuration, defaults to all interfaces.
src/qpid/Plugin.h - added doxygen
src/qpid/Url.cpp,.h - cache string rep, op==, istream/ostream ops.
src/qpid/broker/Broker.h,.cpp - removed getUrl()
src/qpid/cluster/Cluster.h,.cpp - use Url class
src/qpid/cluster/ClusterPlugin.cpp - added --url configuration.
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h
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/cluster/Cluster.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h
incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/Url.cpp
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/Plugin.h Fri Feb 1 08:03:02 2008
@@ -70,15 +70,17 @@
/**
* Initialize Plugin functionality on a Target.
- *
* Plugins should ignore targets they don't recognize.
+ *
+ * Called before the target itself is initialized.
*/
virtual void earlyInitialize(Target&) = 0;
/**
* Initialize Plugin functionality on a Target.
- *
* Plugins should ignore targets they don't recognize.
+ *
+ * Called after the target is fully initialized.
*/
virtual void initialize(Target&) = 0;
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?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/Url.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/Url.cpp Fri Feb 1 08:03:02 2008
@@ -36,10 +36,16 @@
namespace qpid {
-Url Url::getHostnameUrl(uint16_t port) {
+std::ostream& operator<<(std::ostream& os, const TcpAddress& a) {
+ return os << "tcp:" << a.host << ":" << a.port;
+}
+
+std::istream& operator>>(std::istream&, const TcpAddress&);
+
+Url Url::getHostNameUrl(uint16_t port) {
char name[HOST_NAME_MAX];
if (::gethostname(name, sizeof(name)) != 0)
- throw Exception(QPID_MSG("Cannot get host name: " << strError(errno)));
+ throw InvalidUrl(QPID_MSG("Cannot get host name: " <<
strError(errno)));
return Url(TcpAddress(name, port));
}
@@ -66,9 +72,12 @@
}
string Url::str() const {
- ostringstream os;
- os << *this;
- return os.str();
+ if (cache.empty() && !this->empty()) {
+ ostringstream os;
+ os << *this;
+ cache = os.str();
+ }
+ return cache;
}
ostream& operator<<(ostream& os, const Url& url) {
@@ -140,13 +149,22 @@
};
void Url::parse(const char* url) {
+ cache.clear();
if (!boost::spirit::parse(url, UrlGrammar(*this)).full)
throw InvalidUrl(string("Invalid AMQP url: ")+url);
}
void Url::parseNoThrow(const char* url) {
+ cache.clear();
if (!boost::spirit::parse(url, UrlGrammar(*this)).full)
clear();
+}
+
+std::istream& operator>>(std::istream& is, Url& url) {
+ std::string s;
+ is >> s;
+ url.parse(s);
+ return is;
}
} // namespace qpid
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?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/Url.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/Url.h Fri Feb 1 08:03:02 2008
@@ -42,19 +42,16 @@
return y.host==x.host && y.port == x.port;
}
-inline std::ostream& operator<<(std::ostream& os, const TcpAddress& a) {
- return os << "tcp:" << a.host << ":" << a.port;
-}
+std::ostream& operator<<(std::ostream& os, const TcpAddress& a);
-/** Address is a variant of all address types. */
+/** Address is a variant of all address types, more coming in future. */
typedef boost::variant<TcpAddress> Address;
-
/** An AMQP URL contains a list of addresses */
struct Url : public std::vector<Address> {
/** Url with the hostname as returned by gethostname(2) */
- static Url getHostnameUrl(uint16_t port);
+ static Url getHostNameUrl(uint16_t port);
/** Url with local IP address(es), may be more than one address
* on a multi-homed host. */
@@ -79,20 +76,30 @@
/** Parse url, throw InvalidUrl if invalid. */
explicit Url(const char* url) { parse(url); }
+ template<class T> Url& operator=(T s) { parse(s); return *this; }
+
/** Replace contents with parsed URL as defined in
* https://wiki.108.redhat.com/jira/browse/AMQP-95
[EMAIL PROTECTED] InvalidUrl if the url is invalid.
*/
void parse(const char* url);
+ void parse(const std::string& url) { parse(url.c_str()); }
/** Replace contesnts with parsed URL as defined in
* https://wiki.108.redhat.com/jira/browse/AMQP-95
* url.empty() will be true if url is invalid.
*/
void parseNoThrow(const char* url);
+
+ private:
+ mutable std::string cache; // cache string form for efficiency.
};
+inline bool operator==(const Url& a, const Url& b) { return a.str()==b.str(); }
+inline bool operator!=(const Url& a, const Url& b) { return a.str()!=b.str(); }
+
std::ostream& operator<<(std::ostream& os, const Url& url);
+std::istream& operator>>(std::istream& is, Url& url);
} // namespace qpid
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?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Fri Feb 1
08:03:02 2008
@@ -33,7 +33,6 @@
#include "qpid/management/ArgsBrokerEcho.h"
#include "qpid/log/Statement.h"
-#include "qpid/Url.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/framing/ProtocolInitiation.h"
#include "qpid/sys/Acceptor.h"
@@ -61,7 +60,7 @@
Broker::Options::Options(const std::string& name) :
qpid::Options(name),
- port(TcpAddress::DEFAULT_PORT),
+ port(DEFAULT_PORT),
workerThreads(5),
maxConnections(500),
connectionBacklog(10),
@@ -221,10 +220,6 @@
uint16_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) =
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?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h Fri Feb 1 08:03:02
2008
@@ -36,7 +36,6 @@
#include "qpid/management/Broker.h"
#include "qpid/Options.h"
#include "qpid/Plugin.h"
-#include "qpid/Url.h"
#include "qpid/framing/FrameHandler.h"
#include "qpid/framing/OutputHandler.h"
#include "qpid/framing/ProtocolInitiation.h"
@@ -48,6 +47,8 @@
namespace qpid {
namespace broker {
+static const uint16_t DEFAULT_PORT=5672;
+
/**
* A broker instance.
*/
@@ -72,7 +73,7 @@
Broker(const Options& configuration);
static shared_ptr<Broker> create(const Options& configuration);
- static shared_ptr<Broker> create(int16_t port = TcpAddress::DEFAULT_PORT);
+ static shared_ptr<Broker> create(int16_t port = DEFAULT_PORT);
/**
* Return listening port. If called before bind this is
@@ -82,9 +83,6 @@
*/
virtual uint16_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.
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?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Fri Feb 1
08:03:02 2008
@@ -47,7 +47,7 @@
return out;
}
-Cluster::Cluster(const std::string& name_, const std::string& url_,
broker::Broker&) :
+Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker&) :
FrameHandler(0), // FIXME aconway 2008-01-29: handler. +
observer
cpg(*this),
name(name_),
@@ -87,7 +87,7 @@
}
void Cluster::notify() {
- AMQFrame frame(in_place<ClusterNotifyBody>(ProtocolVersion(), url));
+ AMQFrame frame(in_place<ClusterNotifyBody>(ProtocolVersion(), url.str()));
handle(frame);
}
@@ -143,7 +143,7 @@
{
Mutex::ScopedLock l(lock);
members[from].url=notifyIn->getUrl();
- if (!self.id && notifyIn->getUrl() == url)
+ if (!self.id && notifyIn->getUrl() == url.str())
self=from;
lock.notifyAll();
QPID_LOG(trace, *this << ": members joined: " << members);
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?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.h Fri Feb 1
08:03:02 2008
@@ -27,6 +27,8 @@
#include "qpid/sys/Runnable.h"
#include "qpid/sys/Thread.h"
#include "qpid/log/Logger.h"
+#include "qpid/Url.h"
+
#include <boost/optional.hpp>
#include <boost/function.hpp>
@@ -50,8 +52,8 @@
public:
/** Details of a cluster member */
struct Member {
- Member(const std::string& url_=std::string()) : url(url_) {}
- std::string url; ///< Broker address.
+ Member(const Url& url_=Url()) : url(url_) {}
+ Url url; ///< Broker address.
};
typedef std::vector<Member> MemberList;
@@ -61,7 +63,7 @@
* @param name of the cluster.
* @param url of this broker, sent to the cluster.
*/
- Cluster(const std::string& name, const std::string& url, broker::Broker&);
+ Cluster(const std::string& name, const Url& url, broker::Broker&);
virtual ~Cluster();
@@ -115,7 +117,7 @@
mutable sys::Monitor lock;
Cpg cpg;
Cpg::Name name;
- std::string url;
+ Url url;
Id self;
MemberMap members;
sys::Thread dispatcher;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp Fri Feb 1
08:03:02 2008
@@ -15,6 +15,10 @@
* limitations under the License.
*
*/
+#include <boost/program_options/value_semantic.hpp>
+
+
+
#include "qpid/broker/Broker.h"
#include "qpid/cluster/Cluster.h"
#include "qpid/Plugin.h"
@@ -24,21 +28,31 @@
#include <boost/optional.hpp>
#include <boost/utility/in_place_factory.hpp>
+
namespace qpid {
namespace cluster {
using namespace std;
-struct ClusterPlugin : public Plugin {
+struct ClusterOptions : public Options {
+ string name;
+ string url;
+
+ ClusterOptions() : Options("Cluster Options") {
+ addOptions()
+ ("cluster-name", optValue(name, "NAME"), "Name of cluster to join")
+ ("cluster-url", optValue(url,"URL"),
+ "URL of this broker, advertized to the cluster.\n"
+ "Defaults to a URL listing all the local IP addresses\n");
+ }
+
+ Url getUrl(uint16_t port) const {
+ if (url.empty()) return Url::getIpAddressesUrl(port);
+ return Url(url);
+ }
+};
- struct ClusterOptions : public Options {
- string clusterName;
- ClusterOptions() : Options("Cluster Options") {
- addOptions()
- ("cluster", optValue(clusterName, "NAME"),
- "Joins the cluster named NAME");
- }
- };
+struct ClusterPlugin : public Plugin {
ClusterOptions options;
boost::optional<Cluster> cluster;
@@ -50,10 +64,12 @@
void initialize(Plugin::Target& target) {
broker::Broker* broker = dynamic_cast<broker::Broker*>(&target);
// Only provide to a Broker, and only if the --cluster config is set.
- if (broker && !options.clusterName.empty()) {
+ if (broker && !options.name.empty()) {
assert(!cluster); // A process can only belong to one cluster.
- cluster = boost::in_place(options.clusterName, broker->getUrl(),
boost::ref(*broker));
- // broker->add(make_shared_ptr(&cluster->getHandlerUpdater(),
nullDeleter));
+ cluster = boost::in_place(options.name,
+ options.getUrl(broker->getPort()),
+ boost::ref(*broker));
+ // FIXME aconway 2008-02-01: Add observer.
}
}
};
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?rev=617533&r1=617532&r2=617533&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Url.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Url.cpp Fri Feb 1 08:03:02 2008
@@ -33,7 +33,7 @@
url.push_back(TcpAddress("foo.com"));
url.push_back(TcpAddress("bar.com", 6789));
BOOST_CHECK_EQUAL("amqp:tcp:foo.com:5672,tcp:bar.com:6789", url.str());
- BOOST_CHECK_EQUAL("amqp:", Url().str());
+ BOOST_CHECK(Url().str().empty());
}
