You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kg...@apache.org on 2012/05/04 21:45:33 UTC
svn commit: r1334138 [1/2] - in /qpid/trunk/qpid: cpp/src/qpid/broker/
cpp/src/qpid/cluster/ cpp/src/qpid/ha/ cpp/src/tests/
java/broker/src/main/java/org/apache/qpid/qmf/
java/broker/src/main/java/org/apache/qpid/server/configuration/ specs/
tools/src...
Author: kgiusti
Date: Fri May 4 19:45:32 2012
New Revision: 1334138
URL: http://svn.apache.org/viewvc?rev=1334138&view=rev
Log:
QPID-3767: re-index bridge and link by constant name, not address
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Link.h
qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.h
qpid/trunk/qpid/cpp/src/qpid/broker/NameGenerator.h
qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/Backup.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.h
qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
qpid/trunk/qpid/cpp/src/tests/federation.py
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/qmf/QMFService.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/LinkConfigType.java
qpid/trunk/qpid/specs/management-schema.xml
qpid/trunk/qpid/tools/src/py/qpid-tool
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp Fri May 4 19:45:32 2012
@@ -57,22 +57,21 @@ void Bridge::PushHandler::handle(framing
conn->received(frame);
}
-Bridge::Bridge(Link* _link, framing::ChannelId _id, CancellationListener l,
- const _qmf::ArgsLinkBridge& _args,
+Bridge::Bridge(const std::string& _name, Link* _link, framing::ChannelId _id,
+ CancellationListener l, const _qmf::ArgsLinkBridge& _args,
InitializeCallback init) :
- link(_link), id(_id), args(_args), mgmtObject(0),
- listener(l), name(Uuid(true).str()), queueName("qpid.bridge_queue_"), persistenceId(0),
+ link(_link), channel(_id), args(_args), mgmtObject(0),
+ listener(l), name(_name), queueName("qpid.bridge_queue_"), persistenceId(0),
initialize(init), detached(false)
{
- std::stringstream title;
- title << id << "_" << name;
- queueName += title.str();
+ queueName += Uuid(true).str();
ManagementAgent* agent = link->getBroker()->getManagementAgent();
if (agent != 0) {
mgmtObject = new _qmf::Bridge
- (agent, this, link, id, args.i_durable, args.i_src, args.i_dest,
+ (agent, this, link, name, args.i_durable, args.i_src, args.i_dest,
args.i_key, args.i_srcIsQueue, args.i_srcIsLocal,
args.i_tag, args.i_excludes, args.i_dynamic, args.i_sync);
+ mgmtObject->set_channelId(channel);
agent->addObject(mgmtObject);
}
QPID_LOG(debug, "Bridge " << name << " created from " << args.i_src << " to " << args.i_dest);
@@ -90,7 +89,7 @@ void Bridge::create(Connection& c)
conn = &c;
FieldTable options;
if (args.i_sync) options.setInt("qpid.sync_frequency", args.i_sync);
- SessionHandler& sessionHandler = c.getChannel(id);
+ SessionHandler& sessionHandler = c.getChannel(channel);
sessionHandler.setDetachedCallback(
boost::bind(&Bridge::sessionDetached, shared_from_this()));
if (args.i_srcIsLocal) {
@@ -98,15 +97,15 @@ void Bridge::create(Connection& c)
throw Exception("Dynamic routing not supported for push routes");
// Point the bridging commands at the local connection handler
pushHandler.reset(new PushHandler(&c));
- channelHandler.reset(new framing::ChannelHandler(id, pushHandler.get()));
+ channelHandler.reset(new framing::ChannelHandler(channel, pushHandler.get()));
session.reset(new framing::AMQP_ServerProxy::Session(*channelHandler));
peer.reset(new framing::AMQP_ServerProxy(*channelHandler));
- session->attach(name, false);
+ session->attach(queueName, false);
session->commandPoint(0,0);
} else {
- sessionHandler.attachAs(name);
+ sessionHandler.attachAs(queueName);
// Point the bridging commands at the remote peer broker
peer.reset(new framing::AMQP_ServerProxy(sessionHandler.out));
}
@@ -168,6 +167,7 @@ void Bridge::cancel(Connection&)
QPID_LOG(debug, "Cancelled bridge " << name);
}
+/** Notify the bridge that the connection has closed */
void Bridge::closed()
{
if (args.i_dynamic) {
@@ -177,9 +177,10 @@ void Bridge::closed()
QPID_LOG(debug, "Closed bridge " << name);
}
-void Bridge::destroy()
+/** Shut down the bridge */
+void Bridge::close()
{
- listener(this);
+ listener(this); // ask the LinkRegistry to destroy us
}
void Bridge::setPersistenceId(uint64_t pId) const
@@ -187,8 +188,21 @@ void Bridge::setPersistenceId(uint64_t p
persistenceId = pId;
}
+
+const std::string Bridge::ENCODED_IDENTIFIER("bridge.v2");
+const std::string Bridge::ENCODED_IDENTIFIER_V1("bridge");
+
+bool Bridge::isEncodedBridge(const std::string& key)
+{
+ return key == ENCODED_IDENTIFIER || key == ENCODED_IDENTIFIER_V1;
+}
+
+
Bridge::shared_ptr Bridge::decode(LinkRegistry& links, Buffer& buffer)
{
+ string kind;
+ buffer.getShortString(kind);
+
string host;
uint16_t port;
string src;
@@ -196,9 +210,33 @@ Bridge::shared_ptr Bridge::decode(LinkRe
string key;
string id;
string excludes;
+ string name;
+
+ Link::shared_ptr link;
+ if (kind == ENCODED_IDENTIFIER_V1) {
+ /** previous versions identified the bridge by host:port, not by name, and
+ * transport wasn't provided. Try to find a link using those paramters.
+ */
+ buffer.getShortString(host);
+ port = buffer.getShort();
+
+ link = links.getLink(host, port);
+ if (!link) {
+ QPID_LOG(error, "Bridge::decode() failed: cannot find Link for host=" << host << ", port=" << port);
+ return Bridge::shared_ptr();
+ }
+ } else {
+ string linkName;
+
+ buffer.getShortString(name);
+ buffer.getShortString(linkName);
+ link = links.getLink(linkName);
+ if (!link) {
+ QPID_LOG(error, "Bridge::decode() failed: cannot find Link named='" << linkName << "'");
+ return Bridge::shared_ptr();
+ }
+ }
- buffer.getShortString(host);
- port = buffer.getShort();
bool durable(buffer.getOctet());
buffer.getShortString(src);
buffer.getShortString(dest);
@@ -210,15 +248,21 @@ Bridge::shared_ptr Bridge::decode(LinkRe
bool dynamic(buffer.getOctet());
uint16_t sync = buffer.getShort();
- return links.declare(host, port, durable, src, dest, key,
- is_queue, is_local, id, excludes, dynamic, sync).first;
+ if (kind == ENCODED_IDENTIFIER_V1) {
+ /** previous versions did not provide a name for the bridge, so create one
+ */
+ name = createName(link->getName(), src, dest, key);
+ }
+
+ return links.declare(name, *link, durable, src, dest, key, is_queue,
+ is_local, id, excludes, dynamic, sync).first;
}
void Bridge::encode(Buffer& buffer) const
{
- buffer.putShortString(string("bridge"));
- buffer.putShortString(link->getHost());
- buffer.putShort(link->getPort());
+ buffer.putShortString(ENCODED_IDENTIFIER);
+ buffer.putShortString(name);
+ buffer.putShortString(link->getName());
buffer.putOctet(args.i_durable ? 1 : 0);
buffer.putShortString(args.i_src);
buffer.putShortString(args.i_dest);
@@ -233,9 +277,9 @@ void Bridge::encode(Buffer& buffer) cons
uint32_t Bridge::encodedSize() const
{
- return link->getHost().size() + 1 // short-string (host)
- + 7 // short-string ("bridge")
- + 2 // port
+ return ENCODED_IDENTIFIER.size() + 1 // +1 byte length
+ + name.size() + 1
+ + link->getName().size() + 1
+ 1 // durable
+ args.i_src.size() + 1
+ args.i_dest.size() + 1
@@ -259,7 +303,8 @@ management::Manageable::status_t Bridge:
{
if (methodId == _qmf::Bridge::METHOD_CLOSE) {
//notify that we are closed
- destroy();
+ QPID_LOG(debug, "Bridge::close() method called on bridge '" << name << "'");
+ close();
return management::Manageable::STATUS_OK;
} else {
return management::Manageable::STATUS_UNKNOWN_METHOD;
@@ -306,7 +351,7 @@ void Bridge::sendReorigin()
}
bool Bridge::resetProxy()
{
- SessionHandler& sessionHandler = conn->getChannel(id);
+ SessionHandler& sessionHandler = conn->getChannel(channel);
if (!sessionHandler.getSession()) peer.reset();
else peer.reset(new framing::AMQP_ServerProxy(sessionHandler.out));
return peer.get();
@@ -318,7 +363,7 @@ void Bridge::ioThreadPropagateBinding(co
peer->getExchange().bind(queue, exchange, key, args);
} else {
QPID_LOG(error, "Cannot propagate binding for dynamic bridge as session has been detached, deleting dynamic bridge");
- destroy();
+ close();
}
}
@@ -332,9 +377,18 @@ const string& Bridge::getLocalTag() cons
{
return link->getBroker()->getFederationTag();
}
-
void Bridge::sessionDetached() {
detached = true;
}
+std::string Bridge::createName(const std::string& linkName,
+ const std::string& src,
+ const std::string& dest,
+ const std::string& key)
+{
+ std::stringstream keystream;
+ keystream << linkName << "!" << src << "!" << dest << "!" << key;
+ return keystream.str();
+}
+
}}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h Fri May 4 19:45:32 2012
@@ -55,17 +55,18 @@ public:
typedef boost::function<void(Bridge*)> CancellationListener;
typedef boost::function<void(Bridge&, SessionHandler&)> InitializeCallback;
- Bridge(Link* link, framing::ChannelId id, CancellationListener l,
+ Bridge(const std::string& name, Link* link, framing::ChannelId id, CancellationListener l,
const qmf::org::apache::qpid::broker::ArgsLinkBridge& args,
InitializeCallback init
);
~Bridge();
- void create(Connection& c);
- void cancel(Connection& c);
- void closed();
- void destroy();
+ QPID_BROKER_EXTERN void close();
bool isDurable() { return args.i_durable; }
+ Link *getLink() const { return link; }
+ const std::string getSrc() const { return args.i_src; }
+ const std::string getDest() const { return args.i_dest; }
+ const std::string getKey() const { return args.i_key; }
bool isDetached() const { return detached; }
@@ -80,7 +81,11 @@ public:
uint32_t encodedSize() const;
void encode(framing::Buffer& buffer) const;
const std::string& getName() const { return name; }
+
+ static const std::string ENCODED_IDENTIFIER;
+ static const std::string ENCODED_IDENTIFIER_V1;
static Bridge::shared_ptr decode(LinkRegistry& links, framing::Buffer& buffer);
+ static bool isEncodedBridge(const std::string& key);
// Exchange::DynamicBridge methods
void propagateBinding(const std::string& key, const std::string& tagList, const std::string& op, const std::string& origin, qpid::framing::FieldTable* extra_args=0);
@@ -93,6 +98,12 @@ public:
std::string getQueueName() const { return queueName; }
const qmf::org::apache::qpid::broker::ArgsLinkBridge& getArgs() { return args; }
+ /** create a name for a bridge (if none supplied by user config) */
+ static std::string createName(const std::string& linkName,
+ const std::string& src,
+ const std::string& dest,
+ const std::string& key);
+
private:
// Callback when the bridge's session is detached.
void sessionDetached();
@@ -108,8 +119,8 @@ private:
std::auto_ptr<framing::AMQP_ServerProxy::Session> session;
std::auto_ptr<framing::AMQP_ServerProxy> peer;
- Link* link;
- framing::ChannelId id;
+ Link* const link;
+ const framing::ChannelId channel;
qmf::org::apache::qpid::broker::ArgsLinkBridge args;
qmf::org::apache::qpid::broker::Bridge* mgmtObject;
CancellationListener listener;
@@ -121,6 +132,12 @@ private:
InitializeCallback initialize;
bool detached; // Set when session is detached.
bool resetProxy();
+
+ // connection Management (called by owning Link)
+ void create(Connection& c);
+ void cancel(Connection& c);
+ void closed();
+ friend class Link; // to call create, cancel, closed()
};
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Fri May 4 19:45:32 2012
@@ -438,7 +438,7 @@ Manageable* Broker::GetVhostObject(void)
Manageable::status_t Broker::ManagementMethod (uint32_t methodId,
Args& args,
- string&)
+ string& text)
{
Manageable::status_t status = Manageable::STATUS_UNKNOWN_METHOD;
@@ -453,6 +453,14 @@ Manageable::status_t Broker::ManagementM
status = Manageable::STATUS_OK;
break;
case _qmf::Broker::METHOD_CONNECT : {
+ /** Management is creating a Link to a remote broker using the host and port of
+ * the remote. This (old) interface does not allow management to specify a name
+ * for the link, nor does it allow multiple Links to the same remote. Use the
+ * "create()" broker method if these features are needed.
+ * TBD: deprecate this interface.
+ */
+ QPID_LOG(info, "The Broker::connect() method will be removed in a future release of QPID."
+ " Please use the Broker::create() method with type='link' instead.");
_qmf::ArgsBrokerConnect& hp=
dynamic_cast<_qmf::ArgsBrokerConnect&>(args);
@@ -461,13 +469,24 @@ Manageable::status_t Broker::ManagementM
"; durable=" << (hp.i_durable?"T":"F") << "; authMech=\"" << hp.i_authMechanism << "\"");
if (!getProtocolFactory(transport)) {
QPID_LOG(error, "Transport '" << transport << "' not supported");
+ text = "transport type not supported";
return Manageable::STATUS_NOT_IMPLEMENTED;
}
- std::pair<Link::shared_ptr, bool> response =
- links.declare (hp.i_host, hp.i_port, transport, hp.i_durable,
- hp.i_authMechanism, hp.i_username, hp.i_password);
- if (hp.i_durable && response.second)
- store->create(*response.first);
+
+ // Does a link to the remote already exist? If so, re-use the existing link
+ // - this behavior is backward compatible with previous releases.
+ if (!links.getLink(hp.i_host, hp.i_port, transport)) {
+ // new link, need to generate a unique name for it
+ std::pair<Link::shared_ptr, bool> response =
+ links.declare(Link::createName(transport, hp.i_host, hp.i_port),
+ hp.i_host, hp.i_port, transport,
+ hp.i_durable, hp.i_authMechanism, hp.i_username, hp.i_password);
+ if (!response.first) {
+ text = "Unable to create Link";
+ status = Manageable::STATUS_PARAMETER_INVALID;
+ break;
+ }
+ }
status = Manageable::STATUS_OK;
break;
}
@@ -538,6 +557,8 @@ const std::string TYPE_QUEUE("queue");
const std::string TYPE_EXCHANGE("exchange");
const std::string TYPE_TOPIC("topic");
const std::string TYPE_BINDING("binding");
+const std::string TYPE_LINK("link");
+const std::string TYPE_BRIDGE("bridge");
const std::string DURABLE("durable");
const std::string AUTO_DELETE("auto-delete");
const std::string ALTERNATE_EXCHANGE("alternate-exchange");
@@ -549,6 +570,26 @@ const std::string ATTRIBUTE_TIMESTAMP_0_
const std::string _TRUE("true");
const std::string _FALSE("false");
+
+// parameters for creating a Link object, see mgmt schema
+const std::string HOST("host");
+const std::string PORT("port");
+const std::string TRANSPORT("transport");
+const std::string AUTH_MECHANISM("authMechanism");
+const std::string USERNAME("username");
+const std::string PASSWORD("password");
+
+// parameters for creating a Bridge object, see mgmt schema
+const std::string LINK("link");
+const std::string SRC("src");
+const std::string DEST("dest");
+const std::string KEY("key");
+const std::string TAG("tag");
+const std::string EXCLUDES("excludes");
+const std::string SRC_IS_QUEUE("srcIsQueue");
+const std::string SRC_IS_LOCAL("srcIsLocal");
+const std::string DYNAMIC("dynamic");
+const std::string SYNC("sync");
}
struct InvalidBindingIdentifier : public qpid::Exception
@@ -598,6 +639,25 @@ struct UnknownObjectType : public qpid::
std::string getPrefix() const { return "unknown object type"; }
};
+struct ReservedObjectName : public qpid::Exception
+{
+ ReservedObjectName(const std::string& type) : qpid::Exception(type) {}
+ std::string getPrefix() const { return std::string("names prefixed with '")
+ + QPID_NAME_PREFIX + std::string("' are reserved"); }
+};
+
+struct UnsupportedTransport : public qpid::Exception
+{
+ UnsupportedTransport(const std::string& type) : qpid::Exception(type) {}
+ std::string getPrefix() const { return "transport is not supported"; }
+};
+
+struct InvalidParameter : public qpid::Exception
+{
+ InvalidParameter(const std::string& type) : qpid::Exception(type) {}
+ std::string getPrefix() const { return "invalid parameter to method call"; }
+};
+
void Broker::createObject(const std::string& type, const std::string& name,
const Variant::Map& properties, bool /*strict*/, const ConnectionState* context)
{
@@ -669,6 +729,109 @@ void Broker::createObject(const std::str
amqp_0_10::translate(extensions, arguments);
bind(binding.queue, binding.exchange, binding.key, arguments, userId, connectionId);
+
+ } else if (type == TYPE_LINK) {
+
+ QPID_LOG (debug, "createObject: Link; name=" << name << "; args=" << properties );
+
+ if (name.compare(0, QPID_NAME_PREFIX.length(), QPID_NAME_PREFIX) == 0) {
+ QPID_LOG(error, "Link name='" << name << "' cannot use the reserved prefix '" << QPID_NAME_PREFIX << "'");
+ throw ReservedObjectName(name);
+ }
+
+ std::string host;
+ uint16_t port = 0;
+ std::string transport = TCP_TRANSPORT;
+ bool durable = false;
+ std::string authMech, username, password;
+
+ if (!getProtocolFactory(transport)) {
+ QPID_LOG(error, "Transport '" << transport << "' not supported.");
+ throw UnsupportedTransport(transport);
+ }
+
+ for (Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
+ if (i->first == HOST) host = i->second.asString();
+ else if (i->first == PORT) port = i->second.asUint16();
+ else if (i->first == TRANSPORT) transport = i->second.asString();
+ else if (i->first == DURABLE) durable = bool(i->second);
+ else if (i->first == AUTH_MECHANISM) authMech = i->second.asString();
+ else if (i->first == USERNAME) username = i->second.asString();
+ else if (i->first == PASSWORD) password = i->second.asString();
+ else {
+ // TODO: strict checking here
+ }
+ }
+
+ std::pair<boost::shared_ptr<Link>, bool> rc;
+ rc = links.declare(name, host, port, transport, durable, authMech, username, password);
+ if (!rc.first) {
+ QPID_LOG (error, "Failed to create Link object, name=" << name << " remote=" << host << ":" << port <<
+ "; transport=" << transport << "; durable=" << (durable?"T":"F") << "; authMech=\"" << authMech << "\"");
+ throw InvalidParameter(name);
+ }
+ if (!rc.second) {
+ QPID_LOG (error, "Failed to create a new Link object, name=" << name << " already exists.");
+ throw ObjectAlreadyExists(name);
+ }
+
+ } else if (type == TYPE_BRIDGE) {
+
+ QPID_LOG (debug, "createObject: Bridge; name=" << name << "; args=" << properties );
+
+ if (name.compare(0, QPID_NAME_PREFIX.length(), QPID_NAME_PREFIX) == 0) {
+ QPID_LOG(error, "Bridge name='" << name << "' cannot use the reserved prefix '" << QPID_NAME_PREFIX << "'");
+ throw ReservedObjectName(name);
+ }
+
+ std::string linkName;
+ std::string src;
+ std::string dest;
+ std::string key;
+ std::string id;
+ std::string excludes;
+ bool durable = false;
+ bool srcIsQueue = false;
+ bool srcIsLocal = false;
+ bool dynamic = false;
+ uint16_t sync = 0;
+
+ for (Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
+
+ if (i->first == LINK) linkName = i->second.asString();
+ else if (i->first == SRC) src = i->second.asString();
+ else if (i->first == DEST) dest = i->second.asString();
+ else if (i->first == KEY) key = i->second.asString();
+ else if (i->first == TAG) id = i->second.asString();
+ else if (i->first == EXCLUDES) excludes = i->second.asString();
+ else if (i->first == SRC_IS_QUEUE) srcIsQueue = bool(i->second);
+ else if (i->first == SRC_IS_LOCAL) srcIsLocal = bool(i->second);
+ else if (i->first == DYNAMIC) dynamic = bool(i->second);
+ else if (i->first == SYNC) sync = i->second.asUint16();
+ else if (i->first == DURABLE) durable = bool(i->second);
+ else {
+ // TODO: strict checking here
+ }
+ }
+
+ boost::shared_ptr<Link> link;
+ if (linkName.empty() || !(link = links.getLink(linkName))) {
+ QPID_LOG(error, "Link '" << linkName << "' not found; bridge create failed.");
+ throw InvalidParameter(name);
+ }
+ std::pair<Bridge::shared_ptr, bool> rc =
+ links.declare(name, *link, durable, src, dest, key, srcIsQueue, srcIsLocal, id, excludes,
+ dynamic, sync);
+
+ if (!rc.first) {
+ QPID_LOG (error, "Failed to create Bridge object, name=" << name << " link=" << linkName <<
+ "; src=" << src << "; dest=" << dest << "; key=" << key);
+ throw InvalidParameter(name);
+ }
+ if (!rc.second) {
+ QPID_LOG (error, "Failed to create a new Bridge object, name=" << name << " already exists.");
+ throw ObjectAlreadyExists(name);
+ }
} else {
throw UnknownObjectType(type);
}
@@ -691,6 +854,16 @@ void Broker::deleteObject(const std::str
} else if (type == TYPE_BINDING) {
BindingIdentifier binding(name);
unbind(binding.queue, binding.exchange, binding.key, userId, connectionId);
+ } else if (type == TYPE_LINK) {
+ boost::shared_ptr<Link> link = links.getLink(name);
+ if (link) {
+ link->close();
+ }
+ } else if (type == TYPE_BRIDGE) {
+ boost::shared_ptr<Bridge> bridge = links.getBridge(name);
+ if (bridge) {
+ bridge->close();
+ }
} else {
throw UnknownObjectType(type);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Link.cpp Fri May 4 19:45:32 2012
@@ -125,18 +125,19 @@ boost::shared_ptr<Exchange> Link::linkEx
return Exchange::shared_ptr(new LinkExchange(_name));
}
-Link::Link(LinkRegistry* _links,
- MessageStore* _store,
+Link::Link(const string& _name,
+ LinkRegistry* _links,
const string& _host,
uint16_t _port,
const string& _transport,
+ DestroyedListener l,
bool _durable,
const string& _authMechanism,
const string& _username,
const string& _password,
Broker* _broker,
Manageable* parent)
- : links(_links), store(_store),
+ : name(_name), links(_links),
configuredTransport(_transport), configuredHost(_host), configuredPort(_port),
host(_host), port(_port), transport(_transport),
durable(_durable),
@@ -149,6 +150,7 @@ Link::Link(LinkRegistry* _links,
channelCounter(1),
connection(0),
agent(0),
+ listener(l),
timerTask(new LinkTimerTask(*this, broker->getTimer())),
failoverChannel(0)
{
@@ -157,7 +159,10 @@ Link::Link(LinkRegistry* _links,
agent = broker->getManagementAgent();
if (agent != 0)
{
- mgmtObject = new _qmf::Link(agent, this, parent, _host, _port, _transport, _durable);
+ mgmtObject = new _qmf::Link(agent, this, parent, name, durable);
+ mgmtObject->set_host(host);
+ mgmtObject->set_port(port);
+ mgmtObject->set_transport(transport);
agent->addObject(mgmtObject, 0, durable);
}
}
@@ -169,9 +174,9 @@ Link::Link(LinkRegistry* _links,
}
broker->getTimer().add(timerTask);
- stringstream _name;
- _name << "qpid.link." << transport << ":" << host << ":" << port;
- std::pair<Exchange::shared_ptr, bool> rc = broker->getExchanges().declare(_name.str(),
+ stringstream exchangeName;
+ exchangeName << "qpid.link." << name;
+ std::pair<Exchange::shared_ptr, bool> rc = broker->getExchanges().declare(exchangeName.str(),
exchangeTypeName);
failoverExchange = boost::static_pointer_cast<LinkExchange>(rc.first);
assert(failoverExchange);
@@ -245,6 +250,7 @@ void Link::established(Connection* c)
currentInterval = 1;
visitCount = 0;
connection = c;
+
if (closing)
destroy();
else // Process any IO tasks bridges added before established.
@@ -263,7 +269,7 @@ void Link::setUrl(const Url& u) {
namespace {
/** invoked when session used to subscribe to remote's amq.failover exchange detaches */
void sessionDetached(Link *link) {
- QPID_LOG(debug, "detached from 'amq.failover' for link: " << link->getName());
+ QPID_LOG(notice, "detached from 'amq.failover' for link: " << link->getName());
}
}
@@ -271,6 +277,11 @@ namespace {
void Link::opened() {
Mutex::ScopedLock mutex(lock);
if (!connection) return;
+
+ if (!hideManagement() && connection->GetManagementObject()) {
+ mgmtObject->set_connectionRef(connection->GetManagementObject()->getObjectId());
+ }
+
// Get default URL from known-hosts if not already set
if (url.empty()) {
const std::vector<Url>& known = connection->getKnownHosts();
@@ -346,13 +357,14 @@ void Link::closed(int, std::string text)
if (!hideManagement())
mgmtObject->set_lastError (text);
}
+ mgmtObject->set_connectionRef(qpid::management::ObjectId());
}
// Call destroy outside of the lock, don't want to be deleted with lock held.
if (isClosing)
destroy();
}
-// Called in connection IO thread.
+// Called in connection IO thread, cleans up the connection before destroying Link
void Link::destroy ()
{
Bridges toDelete;
@@ -379,9 +391,9 @@ void Link::destroy ()
}
// Now delete all bridges on this link (don't hold the lock for this).
for (Bridges::iterator i = toDelete.begin(); i != toDelete.end(); i++)
- (*i)->destroy();
+ (*i)->close();
toDelete.clear();
- links->destroy (configuredHost, configuredPort);
+ listener(this); // notify LinkRegistry that this Link has been destroyed
}
void Link::add(Bridge::shared_ptr bridge)
@@ -485,12 +497,16 @@ void Link::reconnectLH(const Address& a)
host = a.host;
port = a.port;
transport = a.protocol;
- startConnectionLH();
+
if (!hideManagement()) {
stringstream errorString;
- errorString << "Failed over to " << a;
+ errorString << "Failing over to " << a;
mgmtObject->set_lastError(errorString.str());
+ mgmtObject->set_host(host);
+ mgmtObject->set_port(port);
+ mgmtObject->set_transport(transport);
}
+ startConnectionLH();
}
bool Link::tryFailoverLH() {
@@ -499,15 +515,14 @@ bool Link::tryFailoverLH() {
if (url.empty()) return false;
Address next = url[reconnectNext++];
if (next.host != host || next.port != port || next.protocol != transport) {
- links->changeAddress(Address(transport, host, port), next);
- QPID_LOG(debug, "Inter-broker link failing over to " << next.host << ":" << next.port);
+ QPID_LOG(notice, "Inter-broker link '" << name << "' failing over to " << next);
reconnectLH(next);
return true;
}
return false;
}
-// Management updates for a linke are inconsistent in a cluster, so they are
+// Management updates for a link are inconsistent in a cluster, so they are
// suppressed.
bool Link::hideManagement() const {
return !mgmtObject || ( broker && broker->isInCluster());
@@ -536,18 +551,34 @@ void Link::setPersistenceId(uint64_t id)
const string& Link::getName() const
{
- return configuredHost;
+ return name;
+}
+
+const std::string Link::ENCODED_IDENTIFIER("link.v2");
+const std::string Link::ENCODED_IDENTIFIER_V1("link");
+
+bool Link::isEncodedLink(const std::string& key)
+{
+ return key == ENCODED_IDENTIFIER || key == ENCODED_IDENTIFIER_V1;
}
Link::shared_ptr Link::decode(LinkRegistry& links, Buffer& buffer)
{
+ string kind;
+ buffer.getShortString(kind);
+
string host;
uint16_t port;
string transport;
string authMechanism;
string username;
string password;
+ string name;
+ if (kind == ENCODED_IDENTIFIER) {
+ // newer version provides a link name.
+ buffer.getShortString(name);
+ }
buffer.getShortString(host);
port = buffer.getShort();
buffer.getShortString(transport);
@@ -556,12 +587,21 @@ Link::shared_ptr Link::decode(LinkRegist
buffer.getShortString(username);
buffer.getShortString(password);
- return links.declare(host, port, transport, durable, authMechanism, username, password).first;
+ if (kind == ENCODED_IDENTIFIER_V1) {
+ /** previous versions identified the Link by host:port, there was no name
+ * assigned. So create a name for the new Link.
+ */
+ name = createName(transport, host, port);
+ }
+
+ return links.declare(name, host, port, transport, durable, authMechanism,
+ username, password).first;
}
void Link::encode(Buffer& buffer) const
{
- buffer.putShortString(string("link"));
+ buffer.putShortString(ENCODED_IDENTIFIER);
+ buffer.putShortString(name);
buffer.putShortString(configuredHost);
buffer.putShort(configuredPort);
buffer.putShortString(configuredTransport);
@@ -573,8 +613,9 @@ void Link::encode(Buffer& buffer) const
uint32_t Link::encodedSize() const
{
- return configuredHost.size() + 1 // short-string (host)
- + 5 // short-string ("link")
+ return ENCODED_IDENTIFIER.size() + 1 // +1 byte length
+ + name.size() + 1
+ + configuredHost.size() + 1 // short-string (host)
+ 2 // port
+ configuredTransport.size() + 1 // short-string(transport)
+ 1 // durable
@@ -589,6 +630,7 @@ ManagementObject* Link::GetManagementObj
}
void Link::close() {
+ QPID_LOG(debug, "Link::close(), link=" << name );
Mutex::ScopedLock mutex(lock);
if (!closing) {
closing = true;
@@ -609,36 +651,31 @@ Manageable::status_t Link::ManagementMet
return Manageable::STATUS_OK;
case _qmf::Link::METHOD_BRIDGE :
+ /* TBD: deprecate this interface in favor of the Broker::create() method. The
+ * Broker::create() method allows the user to assign a name to the bridge.
+ */
+ QPID_LOG(info, "The Link::bridge() method will be removed in a future release of QPID."
+ " Please use the Broker::create() method with type='bridge' instead.");
_qmf::ArgsLinkBridge& iargs = (_qmf::ArgsLinkBridge&) args;
- QPID_LOG(debug, "Link::bridge() request received");
+ QPID_LOG(debug, "Link::bridge() request received; src=" << iargs.i_src <<
+ "; dest=" << iargs.i_dest << "; key=" << iargs.i_key);
- // Durable bridges are only valid on durable links
- if (iargs.i_durable && !durable) {
- text = "Can't create a durable route on a non-durable link";
- return Manageable::STATUS_USER;
- }
-
- if (iargs.i_dynamic) {
- Exchange::shared_ptr exchange = getBroker()->getExchanges().get(iargs.i_src);
- if (exchange.get() == 0) {
- text = "Exchange not found";
- return Manageable::STATUS_USER;
- }
- if (!exchange->supportsDynamicBinding()) {
- text = "Exchange type does not support dynamic routing";
- return Manageable::STATUS_USER;
+ // Does a bridge already exist that has the src/dest/key? If so, re-use the
+ // existing bridge - this behavior is backward compatible with previous releases.
+ Bridge::shared_ptr bridge = links->getBridge(*this, iargs.i_src, iargs.i_dest, iargs.i_key);
+ if (!bridge) {
+ // need to create a new bridge on this link.
+ std::pair<Bridge::shared_ptr, bool> rc =
+ links->declare( Bridge::createName(name, iargs.i_src, iargs.i_dest, iargs.i_key),
+ *this, iargs.i_durable,
+ iargs.i_src, iargs.i_dest, iargs.i_key, iargs.i_srcIsQueue,
+ iargs.i_srcIsLocal, iargs.i_tag, iargs.i_excludes,
+ iargs.i_dynamic, iargs.i_sync);
+ if (!rc.first) {
+ text = "invalid parameters";
+ return Manageable::STATUS_PARAMETER_INVALID;
}
}
-
- std::pair<Bridge::shared_ptr, bool> result =
- links->declare (configuredHost, configuredPort, iargs.i_durable, iargs.i_src,
- iargs.i_dest, iargs.i_key, iargs.i_srcIsQueue,
- iargs.i_srcIsLocal, iargs.i_tag, iargs.i_excludes,
- iargs.i_dynamic, iargs.i_sync);
-
- if (result.second && iargs.i_durable)
- store->create(*result.first);
-
return Manageable::STATUS_OK;
}
@@ -716,6 +753,23 @@ void Link::setState(const framing::Field
}
}
+std::string Link::createName(const std::string& transport,
+ const std::string& host,
+ uint16_t port)
+{
+ stringstream linkName;
+ linkName << QPID_NAME_PREFIX << transport << std::string(":")
+ << host << std::string(":") << port;
+ return linkName.str();
+}
+
+
+bool Link::pendingConnection(const std::string& _host, uint16_t _port) const
+{
+ Mutex::ScopedLock mutex(lock);
+ return (isConnecting() && _port == port && _host == host);
+}
+
const std::string Link::exchangeTypeName("qpid.LinkExchange");
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Link.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Link.h?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Link.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Link.h Fri May 4 19:45:32 2012
@@ -25,7 +25,6 @@
#include <boost/shared_ptr.hpp>
#include "qpid/Url.h"
#include "qpid/broker/BrokerImportExport.h"
-#include "qpid/broker/MessageStore.h"
#include "qpid/broker/PersistableConfig.h"
#include "qpid/broker/Bridge.h"
#include "qpid/broker/BrokerImportExport.h"
@@ -52,8 +51,8 @@ class LinkExchange;
class Link : public PersistableConfig, public management::Manageable {
private:
mutable sys::Mutex lock;
+ const std::string name;
LinkRegistry* links;
- MessageStore* store;
// these remain constant across failover - used to identify this link
const std::string configuredTransport;
@@ -85,6 +84,7 @@ class Link : public PersistableConfig, p
uint channelCounter;
Connection* connection;
management::ManagementAgent* agent;
+ boost::function<void(Link*)> listener;
boost::intrusive_ptr<sys::TimerTask> timerTask;
boost::shared_ptr<broker::LinkExchange> failoverExchange; // subscribed to remote's amq.failover exchange
uint failoverChannel;
@@ -101,27 +101,32 @@ class Link : public PersistableConfig, p
void setStateLH (int newState);
void startConnectionLH(); // Start the IO Connection
- void destroy(); // Called when mgmt deletes this link
+ void destroy(); // Cleanup connection before link goes away
void ioThreadProcessing(); // Called on connection's IO thread by request
bool tryFailoverLH(); // Called during maintenance visit
bool hideManagement() const;
+ void reconnectLH(const Address&); //called by LinkRegistry
- void established(Connection*); // Called when connection is create
+ // connection management (called by LinkRegistry)
+ void established(Connection*); // Called when connection is created
void opened(); // Called when connection is open (after create)
void closed(int, std::string); // Called when connection goes away
- void reconnectLH(const Address&); //called by LinkRegistry
+ void notifyConnectionForced(const std::string text);
void closeConnection(const std::string& reason);
+ bool pendingConnection(const std::string& host, uint16_t port) const; // is Link trying to connect to this remote?
friend class LinkRegistry; // to call established, opened, closed
public:
typedef boost::shared_ptr<Link> shared_ptr;
+ typedef boost::function<void(Link*)> DestroyedListener;
- Link(LinkRegistry* links,
- MessageStore* store,
+ Link(const std::string& name,
+ LinkRegistry* links,
const std::string& host,
uint16_t port,
const std::string& transport,
+ DestroyedListener l,
bool durable,
const std::string& authMechanism,
const std::string& username,
@@ -148,15 +153,17 @@ class Link : public PersistableConfig, p
void cancel(Bridge::shared_ptr);
QPID_BROKER_EXTERN void setUrl(const Url&); // Set URL for reconnection.
- QPID_BROKER_EXTERN void close(); // Close the link from within the broker.
+
+ // Close the link.
+ QPID_BROKER_EXTERN void close();
std::string getAuthMechanism() { return authMechanism; }
std::string getUsername() { return username; }
std::string getPassword() { return password; }
Broker* getBroker() { return broker; }
- void notifyConnectionForced(const std::string text);
void setPassive(bool p);
+ bool isConnecting() const { return state == STATE_CONNECTING; }
// PersistableConfig:
void setPersistenceId(uint64_t id) const;
@@ -165,7 +172,10 @@ class Link : public PersistableConfig, p
void encode(framing::Buffer& buffer) const;
const std::string& getName() const;
+ static const std::string ENCODED_IDENTIFIER;
+ static const std::string ENCODED_IDENTIFIER_V1;
static Link::shared_ptr decode(LinkRegistry& links, framing::Buffer& buffer);
+ static bool isEncodedLink(const std::string& key);
// Manageable entry points
management::ManagementObject* GetManagementObject(void) const;
@@ -178,6 +188,11 @@ class Link : public PersistableConfig, p
// replicate internal state of this Link for clustering
void getState(framing::FieldTable& state) const;
void setState(const framing::FieldTable& state);
+
+ /** create a name for a link (if none supplied by user config) */
+ static std::string createName(const std::string& transport,
+ const std::string& host,
+ uint16_t port);
};
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.cpp Fri May 4 19:45:32 2012
@@ -68,27 +68,34 @@ LinkRegistry::LinkRegistry (Broker* _bro
LinkRegistry::~LinkRegistry() {}
+/** find link by the *configured* remote address */
+boost::shared_ptr<Link> LinkRegistry::getLink(const std::string& host,
+ uint16_t port,
+ const std::string& transport)
+{
+ Mutex::ScopedLock locker(lock);
+ for (LinkMap::iterator i = links.begin(); i != links.end(); ++i) {
+ Link::shared_ptr& link = i->second;
+ if (link->getHost() == host &&
+ link->getPort() == port &&
+ (transport.empty() || link->getTransport() == transport))
+ return link;
+ }
+ return boost::shared_ptr<Link>();
+}
-void LinkRegistry::changeAddress(const qpid::Address& oldAddress, const qpid::Address& newAddress)
+/** find link by name */
+boost::shared_ptr<Link> LinkRegistry::getLink(const std::string& name)
{
Mutex::ScopedLock locker(lock);
- std::string oldKey = createKey(oldAddress);
- std::string newKey = createKey(newAddress);
- if (links.find(newKey) != links.end()) {
- QPID_LOG(error, "Attempted to update key from " << oldKey << " to " << newKey << " which is already in use");
- } else {
- LinkMap::iterator i = links.find(oldKey);
- if (i == links.end()) {
- QPID_LOG(error, "Attempted to update key from " << oldKey << " which does not exist, to " << newKey);
- } else {
- links[newKey] = i->second;
- links.erase(oldKey);
- QPID_LOG(info, "Updated link key from " << oldKey << " to " << newKey);
- }
- }
+ LinkMap::iterator l = links.find(name);
+ if (l != links.end())
+ return l->second;
+ return boost::shared_ptr<Link>();
}
-pair<Link::shared_ptr, bool> LinkRegistry::declare(const string& host,
+pair<Link::shared_ptr, bool> LinkRegistry::declare(const string& name,
+ const string& host,
uint16_t port,
const string& transport,
bool durable,
@@ -98,24 +105,53 @@ pair<Link::shared_ptr, bool> LinkRegistr
{
Mutex::ScopedLock locker(lock);
- string key = createKey(host, port);
- LinkMap::iterator i = links.find(key);
+ LinkMap::iterator i = links.find(name);
if (i == links.end())
{
Link::shared_ptr link;
- link = Link::shared_ptr (new Link (this, store, host, port, transport, durable,
- authMechanism, username, password,
- broker, parent));
- links[key] = link;
+ link = Link::shared_ptr (new Link (name, this, host, port, transport,
+ boost::bind(&LinkRegistry::linkDestroyed, this, _1),
+ durable, authMechanism, username, password, broker,
+ parent));
+ if (durable && store) store->create(*link);
+ links[name] = link;
+ QPID_LOG(debug, "Creating new link; name=" << name );
return std::pair<Link::shared_ptr, bool>(link, true);
}
return std::pair<Link::shared_ptr, bool>(i->second, false);
}
-pair<Bridge::shared_ptr, bool> LinkRegistry::declare(const std::string& host,
- uint16_t port,
+/** find bridge by link & route info */
+Bridge::shared_ptr LinkRegistry::getBridge(const Link& link,
+ const std::string& src,
+ const std::string& dest,
+ const std::string& key)
+{
+ Mutex::ScopedLock locker(lock);
+ for (BridgeMap::iterator i = bridges.begin(); i != bridges.end(); ++i) {
+ if (i->second->getSrc() == src && i->second->getDest() == dest &&
+ i->second->getKey() == key && i->second->getLink() &&
+ i->second->getLink()->getName() == link.getName()) {
+ return i->second;
+ }
+ }
+ return Bridge::shared_ptr();
+}
+
+/** find bridge by name */
+Bridge::shared_ptr LinkRegistry::getBridge(const std::string& name)
+{
+ Mutex::ScopedLock locker(lock);
+ BridgeMap::iterator b = bridges.find(name);
+ if (b != bridges.end())
+ return b->second;
+ return Bridge::shared_ptr();
+}
+
+pair<Bridge::shared_ptr, bool> LinkRegistry::declare(const std::string& name,
+ Link& link,
bool durable,
const std::string& src,
const std::string& dest,
@@ -130,18 +166,26 @@ pair<Bridge::shared_ptr, bool> LinkRegis
)
{
Mutex::ScopedLock locker(lock);
- QPID_LOG(debug, "Bridge declared " << host << ": " << port << " from " << src << " to " << dest << " (" << key << ")");
- string linkKey = createKey(host, port);
- stringstream keystream;
- keystream << linkKey << "!" << src << "!" << dest << "!" << key;
- string bridgeKey = keystream.str();
-
- LinkMap::iterator l = links.find(linkKey);
- if (l == links.end())
- return pair<Bridge::shared_ptr, bool>(Bridge::shared_ptr(), false);
+ // Durable bridges are only valid on durable links
+ if (durable && !link.isDurable()) {
+ QPID_LOG(error, "Can't create a durable route '" << name << "' on a non-durable link '" << link.getName());
+ return pair<Bridge::shared_ptr, bool>(Bridge::shared_ptr(), false);
+ }
- BridgeMap::iterator b = bridges.find(bridgeKey);
+ if (dynamic) {
+ Exchange::shared_ptr exchange = broker->getExchanges().get(src);
+ if (exchange.get() == 0) {
+ QPID_LOG(error, "Exchange not found, name='" << src << "'" );
+ return pair<Bridge::shared_ptr, bool>(Bridge::shared_ptr(), false);
+ }
+ if (!exchange->supportsDynamicBinding()) {
+ QPID_LOG(error, "Exchange type does not support dynamic routing, name='" << src << "'");
+ return pair<Bridge::shared_ptr, bool>(Bridge::shared_ptr(), false);
+ }
+ }
+
+ BridgeMap::iterator b = bridges.find(name);
if (b == bridges.end())
{
_qmf::ArgsLinkBridge args;
@@ -159,23 +203,29 @@ pair<Bridge::shared_ptr, bool> LinkRegis
args.i_sync = sync;
bridge = Bridge::shared_ptr
- (new Bridge (l->second.get(), l->second->nextChannel(),
- boost::bind(&LinkRegistry::destroy, this,
- host, port, src, dest, key),
- args, init));
- bridges[bridgeKey] = bridge;
- l->second->add(bridge);
+ (new Bridge (name, &link, link.nextChannel(),
+ boost::bind(&LinkRegistry::destroyBridge, this, _1),
+ args, init));
+ bridges[name] = bridge;
+ link.add(bridge);
+ if (durable && store)
+ store->create(*bridge);
+
+ QPID_LOG(debug, "Bridge '" << name <<"' declared on link '" << link.getName() <<
+ "' from " << src << " to " << dest << " (" << key << ")");
+
return std::pair<Bridge::shared_ptr, bool>(bridge, true);
}
return std::pair<Bridge::shared_ptr, bool>(b->second, false);
}
-void LinkRegistry::destroy(const string& host, const uint16_t port)
+/** called back by the link when it has completed its cleanup and can be removed. */
+void LinkRegistry::linkDestroyed(Link *link)
{
+ QPID_LOG(debug, "LinkRegistry::destroy(); link= " << link->getName());
Mutex::ScopedLock locker(lock);
- string key = createKey(host, port);
- LinkMap::iterator i = links.find(key);
+ LinkMap::iterator i = links.find(link->getName());
if (i != links.end())
{
if (i->second->isDurable() && store)
@@ -184,27 +234,20 @@ void LinkRegistry::destroy(const string&
}
}
-void LinkRegistry::destroy(const std::string& host,
- const uint16_t port,
- const std::string& src,
- const std::string& dest,
- const std::string& key)
+/** called back by bridge when its destruction has been requested */
+void LinkRegistry::destroyBridge(Bridge *bridge)
{
+ QPID_LOG(debug, "LinkRegistry::destroy(); bridge= " << bridge->getName());
Mutex::ScopedLock locker(lock);
- string linkKey = createKey(host, port);
- stringstream keystream;
- keystream << linkKey << "!" << src << "!" << dest << "!" << key;
- string bridgeKey = keystream.str();
- LinkMap::iterator l = links.find(linkKey);
- if (l == links.end())
- return;
-
- BridgeMap::iterator b = bridges.find(bridgeKey);
+ BridgeMap::iterator b = bridges.find(bridge->getName());
if (b == bridges.end())
return;
- l->second->cancel(b->second);
+ Link *link = b->second->getLink();
+ if (link) {
+ link->cancel(b->second);
+ }
if (b->second->isDurable())
store->destroy(*(b->second));
bridges.erase(b);
@@ -219,28 +262,73 @@ MessageStore* LinkRegistry::getStore() c
return store;
}
-Link::shared_ptr LinkRegistry::findLink(const std::string& keyOrMgmtId)
-{
- // Convert keyOrMgmtId to a host:port key.
- //
- // TODO aconway 2011-02-01: centralize code that constructs/parses
- // connection management IDs. Currently sys:: protocol factories
- // and IO plugins construct the IDs and LinkRegistry parses them.
- size_t separator = keyOrMgmtId.find('-');
- if (separator == std::string::npos) separator = 0;
- std::string key = keyOrMgmtId.substr(separator+1, std::string::npos);
+namespace {
+ void extractHostPort(const std::string& connId, std::string *host, uint16_t *port)
+ {
+ // Extract host and port of remote broker from connection id string.
+ //
+ // TODO aconway 2011-02-01: centralize code that constructs/parses connection
+ // management IDs. Currently sys:: protocol factories and IO plugins construct the
+ // IDs and LinkRegistry parses them.
+ // KAG: current connection id format assumed:
+ // "localhost:port-remotehost:port". In the case of IpV6, the host addresses are
+ // contained within brackets "[...]", example:
+ // connId="[::1]:36859-[::1]:48603". Liberal use of "asserts" provided to alert us
+ // if this assumption changes!
+ size_t separator = connId.find('-');
+ assert(separator != std::string::npos);
+ std::string remote = connId.substr(separator+1, std::string::npos);
+ separator = remote.rfind(":");
+ assert(separator != std::string::npos);
+ *host = remote.substr(0, separator);
+ // IPv6 - host is bracketed by "[]", strip them
+ if ((*host)[0] == '[' && (*host)[host->length() - 1] == ']') {
+ *host = host->substr(1, host->length() - 2);
+ }
+ try {
+ *port = boost::lexical_cast<uint16_t>(remote.substr(separator+1, std::string::npos));
+ } catch (const boost::bad_lexical_cast&) {
+ QPID_LOG(error, "Invalid format for connection identifier! '" << connId << "'");
+ assert(false);
+ }
+ }
+}
+/** find the Link that corresponds to the given connection */
+Link::shared_ptr LinkRegistry::findLink(const std::string& connId)
+{
Mutex::ScopedLock locker(lock);
- LinkMap::iterator l = links.find(key);
- if (l != links.end()) return l->second;
- else return Link::shared_ptr();
+ ConnectionMap::iterator c = connections.find(connId);
+ if (c != connections.end()) {
+ LinkMap::iterator l = links.find(c->second);
+ if (l != links.end())
+ return l->second;
+ }
+ return Link::shared_ptr();
}
void LinkRegistry::notifyConnection(const std::string& key, Connection* c)
{
- Link::shared_ptr link = findLink(key);
+ // find a link that is attempting to connect to the remote, and
+ // create a mapping from connection id to link
+ QPID_LOG(debug, "LinkRegistry::notifyConnection(); key=" << key );
+ std::string host;
+ uint16_t port;
+ extractHostPort( key, &host, &port );
+ Link::shared_ptr link;
+ {
+ Mutex::ScopedLock locker(lock);
+ for (LinkMap::iterator l = links.begin(); l != links.end(); ++l) {
+ if (l->second->pendingConnection(host, port)) {
+ link = l->second;
+ connections[key] = link->getName();
+ link->established(c);
+ break;
+ }
+ }
+ }
+
if (link) {
- link->established(c);
c->setUserId(str(format("%1%@%2%") % link->getUsername() % realm));
}
}
@@ -343,20 +431,6 @@ std::string LinkRegistry::getAuthIdentit
}
-std::string LinkRegistry::createKey(const qpid::Address& a) {
- // TODO aconway 2010-05-11: key should also include protocol/transport to
- // be unique. Requires refactor of LinkRegistry interface.
- return createKey(a.host, a.port);
-}
-
-std::string LinkRegistry::createKey(const std::string& host, uint16_t port) {
- // TODO aconway 2010-05-11: key should also include protocol/transport to
- // be unique. Requires refactor of LinkRegistry interface.
- stringstream keystream;
- keystream << host << ":" << port;
- return keystream.str();
-}
-
void LinkRegistry::setPassive(bool p)
{
Mutex::ScopedLock locker(lock);
@@ -369,10 +443,12 @@ void LinkRegistry::setPassive(bool p)
}
void LinkRegistry::eachLink(boost::function<void(boost::shared_ptr<Link>)> f) {
+ Mutex::ScopedLock locker(lock);
for (LinkMap::iterator i = links.begin(); i != links.end(); ++i) f(i->second);
}
void LinkRegistry::eachBridge(boost::function<void(boost::shared_ptr<Bridge>)> f) {
+ Mutex::ScopedLock locker(lock);
for (BridgeMap::iterator i = bridges.begin(); i != bridges.end(); ++i) f(i->second);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.h?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/LinkRegistry.h Fri May 4 19:45:32 2012
@@ -42,9 +42,11 @@ namespace broker {
class LinkRegistry {
typedef std::map<std::string, boost::shared_ptr<Link> > LinkMap;
typedef std::map<std::string, Bridge::shared_ptr> BridgeMap;
+ typedef std::map<std::string, std::string> ConnectionMap;
- LinkMap links;
- BridgeMap bridges;
+ LinkMap links; /** indexed by name of Link */
+ BridgeMap bridges; /** indexed by name of Bridge */
+ ConnectionMap connections; /** indexed by connection identifier, gives link name */
qpid::sys::Mutex lock;
Broker* broker;
@@ -54,15 +56,18 @@ namespace broker {
std::string realm;
boost::shared_ptr<Link> findLink(const std::string& key);
- static std::string createKey(const Address& address);
- static std::string createKey(const std::string& host, uint16_t port);
- // Methods called by the connection observer.
+ // Methods called by the connection observer, key is connection identifier
void notifyConnection (const std::string& key, Connection* c);
void notifyOpened (const std::string& key);
void notifyClosed (const std::string& key);
void notifyConnectionForced (const std::string& key, const std::string& text);
- friend class LinkRegistryConnectionObserver;
+ friend class LinkRegistryConnectionObserver;
+
+ /** Notify the registry that a Link has been destroyed */
+ void linkDestroyed(Link*);
+ /** Request to destroy a Bridge */
+ void destroyBridge(Bridge*);
public:
QPID_BROKER_EXTERN LinkRegistry (); // Only used in store tests
@@ -70,17 +75,27 @@ namespace broker {
QPID_BROKER_EXTERN ~LinkRegistry();
QPID_BROKER_EXTERN std::pair<boost::shared_ptr<Link>, bool>
- declare(const std::string& host,
+ declare(const std::string& name,
+ const std::string& host,
uint16_t port,
const std::string& transport,
bool durable,
const std::string& authMechanism,
const std::string& username,
const std::string& password);
+ /** determine if Link exists */
+ QPID_BROKER_EXTERN boost::shared_ptr<Link>
+ getLink(const std::string& name);
+ /** host,port,transport will be matched against the configured values, which may
+ be different from the current values due to failover */
+ QPID_BROKER_EXTERN boost::shared_ptr<Link>
+ getLink(const std::string& configHost,
+ uint16_t configPort,
+ const std::string& configTransport = std::string());
QPID_BROKER_EXTERN std::pair<Bridge::shared_ptr, bool>
- declare(const std::string& host,
- uint16_t port,
+ declare(const std::string& name,
+ Link& link,
bool durable,
const std::string& src,
const std::string& dest,
@@ -93,14 +108,14 @@ namespace broker {
uint16_t sync,
Bridge::InitializeCallback=0
);
-
- QPID_BROKER_EXTERN void destroy(const std::string& host, const uint16_t port);
-
- QPID_BROKER_EXTERN void destroy(const std::string& host,
- const uint16_t port,
- const std::string& src,
- const std::string& dest,
- const std::string& key);
+ /** determine if Bridge exists */
+ QPID_BROKER_EXTERN Bridge::shared_ptr
+ getBridge(const std::string& name);
+ QPID_BROKER_EXTERN Bridge::shared_ptr
+ getBridge(const Link& link,
+ const std::string& src,
+ const std::string& dest,
+ const std::string& key);
/**
* Register the manageable parent for declared queues
@@ -126,11 +141,6 @@ namespace broker {
QPID_BROKER_EXTERN uint16_t getPort (const std::string& key);
/**
- * Called by links failing over to new address
- */
- void changeAddress(const Address& oldAddress, const Address& newAddress);
-
- /**
* Called to alter passive state. In passive state the links
* and bridges managed by a link registry will be recorded and
* updated but links won't actually establish connections and
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/NameGenerator.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/NameGenerator.h?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/NameGenerator.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/NameGenerator.h Fri May 4 19:45:32 2012
@@ -32,6 +32,7 @@ namespace qpid {
NameGenerator(const std::string& base);
std::string generate();
};
+ const std::string QPID_NAME_PREFIX("qpid."); // reserved for private names
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp Fri May 4 19:45:32 2012
@@ -144,11 +144,13 @@ RecoverableTransaction::shared_ptr Recov
RecoverableConfig::shared_ptr RecoveryManagerImpl::recoverConfig(framing::Buffer& buffer)
{
string kind;
-
+ uint32_t p = buffer.getPosition();
buffer.getShortString (kind);
- if (kind == "link")
+ buffer.setPosition(p);
+
+ if (Link::isEncodedLink(kind))
return RecoverableConfig::shared_ptr(new RecoverableConfigImpl(Link::decode (links, buffer)));
- else if (kind == "bridge")
+ else if (Bridge::isEncodedBridge(kind))
return RecoverableConfig::shared_ptr(new RecoverableConfigImpl(Bridge::decode (links, buffer)));
return RecoverableConfig::shared_ptr(); // TODO: raise an exception instead
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Fri May 4 19:45:32 2012
@@ -782,16 +782,18 @@ void Connection::managementSetupState(
void Connection::config(const std::string& encoded) {
Buffer buf(const_cast<char*>(encoded.data()), encoded.size());
string kind;
+ uint32_t p = buf.getPosition();
buf.getShortString (kind);
- if (kind == "link") {
+ buf.setPosition(p);
+ if (broker::Link::isEncodedLink(kind)) {
broker::Link::shared_ptr link =
- broker::Link::decode(cluster.getBroker().getLinks(), buf);
+ broker::Link::decode(cluster.getBroker().getLinks(), buf);
QPID_LOG(debug, cluster << " updated link "
<< link->getHost() << ":" << link->getPort());
}
- else if (kind == "bridge") {
+ else if (broker::Bridge::isEncodedBridge(kind)) {
broker::Bridge::shared_ptr bridge =
- broker::Bridge::decode(cluster.getBroker().getLinks(), buf);
+ broker::Bridge::decode(cluster.getBroker().getLinks(), buf);
QPID_LOG(debug, cluster << " updated bridge " << bridge->getName());
}
else throw Exception(QPID_MSG("Update failed, invalid kind of config: " << kind));
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/Backup.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/Backup.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/Backup.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/Backup.cpp Fri May 4 19:45:32 2012
@@ -57,8 +57,10 @@ void Backup::initialize(const Url& url)
if (url.empty()) throw Url::Invalid("HA broker URL is empty");
QPID_LOG(notice, "HA: Backup initialized: " << url);
string protocol = url[0].protocol.empty() ? "tcp" : url[0].protocol;
+ framing::Uuid uuid(true);
// Declare the link
std::pair<Link::shared_ptr, bool> result = broker.getLinks().declare(
+ broker::QPID_NAME_PREFIX + string("ha.link.") + uuid.str(),
url[0].host, url[0].port, protocol,
false, // durable
settings.mechanism, settings.username, settings.password);
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp Fri May 4 19:45:32 2012
@@ -190,8 +190,11 @@ BrokerReplicator::BrokerReplicator(HaBro
: Exchange(QPID_CONFIGURATION_REPLICATOR),
haBroker(hb), broker(hb.getBroker()), link(l)
{
+ framing::Uuid uuid(true);
+ const std::string name(QPID_CONFIGURATION_REPLICATOR + ".bridge." + uuid.str());
broker.getLinks().declare(
- link->getHost(), link->getPort(),
+ name, // name for bridge
+ *link, // parent
false, // durable
QPID_CONFIGURATION_REPLICATOR, // src
QPID_CONFIGURATION_REPLICATOR, // dest
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/HaBroker.cpp Fri May 4 19:45:32 2012
@@ -119,7 +119,9 @@ Manageable::status_t HaBroker::Managemen
boost::shared_ptr<broker::Queue> queue = broker.getQueues().get(bq_args.i_queue);
Url url(bq_args.i_broker);
string protocol = url[0].protocol.empty() ? "tcp" : url[0].protocol;
+ framing::Uuid uuid(true);
std::pair<broker::Link::shared_ptr, bool> result = broker.getLinks().declare(
+ broker::QPID_NAME_PREFIX + string("ha.link.") + uuid.str(),
url[0].host, url[0].port, protocol,
false, // durable
settings.mechanism, settings.username, settings.password);
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.cpp Fri May 4 19:45:32 2012
@@ -60,9 +60,11 @@ QueueReplicator::QueueReplicator(boost::
// This must be separate from the constructor so we can call shared_from_this.
void QueueReplicator::activate() {
- // Note this may create a new bridge or use an existing one.
+ sys::Mutex::ScopedLock l(lock);
+ std::pair<Bridge::shared_ptr, bool> result =
queue->getBroker()->getLinks().declare(
- link->getHost(), link->getPort(),
+ bridgeName,
+ *link,
false, // durable
queue->getName(), // src
getName(), // dest
@@ -77,21 +79,24 @@ void QueueReplicator::activate() {
// before initializeBridge is called.
boost::bind(&QueueReplicator::initializeBridge, shared_from_this(), _1, _2)
);
+ bridge = result.first;
}
QueueReplicator::~QueueReplicator() {}
void QueueReplicator::deactivate() {
+ // destroy the route
sys::Mutex::ScopedLock l(lock);
- queue->getBroker()->getLinks().destroy(
- link->getHost(), link->getPort(), queue->getName(), getName(), string());
- QPID_LOG(debug, logPrefix << "Deactivated bridge " << bridgeName);
+ if (bridge) {
+ bridge->close();
+ bridge.reset();
+ QPID_LOG(debug, logPrefix << "Deactivated bridge " << bridgeName);
+ }
}
// Called in a broker connection thread when the bridge is created.
void QueueReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHandler) {
sys::Mutex::ScopedLock l(lock);
- bridgeName = bridge.getName();
framing::AMQP_ServerProxy peer(sessionHandler.out);
const qmf::org::apache::qpid::broker::ArgsLinkBridge& args(bridge.getArgs());
framing::FieldTable settings;
Modified: qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.h?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/ha/QueueReplicator.h Fri May 4 19:45:32 2012
@@ -78,6 +78,7 @@ class QueueReplicator : public broker::E
sys::Mutex lock;
boost::shared_ptr<broker::Queue> queue;
boost::shared_ptr<broker::Link> link;
+ boost::shared_ptr<broker::Bridge> bridge;
};
}} // namespace qpid::ha
Modified: qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_tests.py?rev=1334138&r1=1334137&r2=1334138&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_tests.py Fri May 4 19:45:32 2012
@@ -768,6 +768,35 @@ acl deny all all
fetch(cluster[2])
+ def _verify_federation(self, src_broker, src, dst_broker, dst, timeout=30):
+ """ Prove that traffic can pass between two federated brokers.
+ """
+ tot_time = 0
+ active = False
+ send_session = src_broker.connect().session()
+ sender = send_session.sender(src)
+ receive_session = dst_broker.connect().session()
+ receiver = receive_session.receiver(dst)
+ while not active and tot_time < timeout:
+ sender.send(Message("Hello from Source!"))
+ try:
+ receiver.fetch(timeout = 1)
+ receive_session.acknowledge()
+ # Get this far without Empty exception, and the link is good!
+ active = True
+ while True:
+ # Keep receiving msgs, as several may have accumulated
+ receiver.fetch(timeout = 1)
+ receive_session.acknowledge()
+ except Empty:
+ if not active:
+ tot_time += 1
+ receiver.close()
+ receive_session.close()
+ sender.close()
+ send_session.close()
+ return active
+
def test_federation_failover(self):
"""
Verify that federation operates across failures occuring in a cluster.
@@ -778,38 +807,6 @@ acl deny all all
cluster to newly-added members
"""
- TIMEOUT = 30
- def verify(src_broker, src, dst_broker, dst, timeout=TIMEOUT):
- """ Prove that traffic can pass from source fed broker to
- destination fed broker
- """
- tot_time = 0
- active = False
- send_session = src_broker.connect().session()
- sender = send_session.sender(src)
- receive_session = dst_broker.connect().session()
- receiver = receive_session.receiver(dst)
- while not active and tot_time < timeout:
- sender.send(Message("Hello from Source!"))
- try:
- receiver.fetch(timeout = 1)
- receive_session.acknowledge()
- # Get this far without Empty exception, and the link is good!
- active = True
- while True:
- # Keep receiving msgs, as several may have accumulated
- receiver.fetch(timeout = 1)
- receive_session.acknowledge()
- except Empty:
- if not active:
- tot_time += 1
- receiver.close()
- receive_session.close()
- sender.close()
- send_session.close()
- self.assertTrue(active, "Bridge failed to become active")
-
-
# 2 node cluster source, 2 node cluster destination
src_cluster = self.cluster(2, expect=EXPECT_EXIT_FAIL)
src_cluster.ready();
@@ -848,43 +845,145 @@ acl deny all all
self.assertEqual(result.status, 0, result)
# check that traffic passes
- verify(src_cluster[0], "srcQ", dst_cluster[0], "destQ")
+ assert self._verify_federation(src_cluster[0], "srcQ", dst_cluster[0], "destQ")
# add src[2] broker to source cluster
src_cluster.start(expect=EXPECT_EXIT_FAIL);
src_cluster.ready();
- verify(src_cluster[2], "srcQ", dst_cluster[0], "destQ")
+ assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[0], "destQ")
# Kill src[0]. dst[0] should fail over to src[1]
src_cluster[0].kill()
for b in src_cluster[1:]: b.ready()
- verify(src_cluster[1], "srcQ", dst_cluster[0], "destQ")
+ assert self._verify_federation(src_cluster[1], "srcQ", dst_cluster[0], "destQ")
# Kill src[1], dst[0] should fail over to src[2]
src_cluster[1].kill()
for b in src_cluster[2:]: b.ready()
- verify(src_cluster[2], "srcQ", dst_cluster[0], "destQ")
+ assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[0], "destQ")
# Kill dest[0], force failover to dest[1]
dst_cluster[0].kill()
for b in dst_cluster[1:]: b.ready()
- verify(src_cluster[2], "srcQ", dst_cluster[1], "destQ")
+ assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[1], "destQ")
# Add dest[2]
# dest[1] syncs dest[2] to current remote state
dst_cluster.start(expect=EXPECT_EXIT_FAIL);
for b in dst_cluster[1:]: b.ready()
- verify(src_cluster[2], "srcQ", dst_cluster[1], "destQ")
+ assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[1], "destQ")
# Kill dest[1], force failover to dest[2]
dst_cluster[1].kill()
for b in dst_cluster[2:]: b.ready()
- verify(src_cluster[2], "srcQ", dst_cluster[2], "destQ")
+ assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[2], "destQ")
for i in range(2, len(src_cluster)): src_cluster[i].kill()
for i in range(2, len(dst_cluster)): dst_cluster[i].kill()
+ def test_federation_multilink_failover(self):
+ """
+ Verify that multi-link federation operates across failures occuring in
+ a cluster.
+ """
+
+ # 1 node cluster source, 1 node cluster destination
+ src_cluster = self.cluster(1, expect=EXPECT_EXIT_FAIL)
+ src_cluster.ready();
+ dst_cluster = self.cluster(1, expect=EXPECT_EXIT_FAIL)
+ dst_cluster.ready();
+
+ # federate a direct binding across two separate links
+
+ # first, create a direct exchange bound to two queues using different
+ # bindings
+ cmd = self.popen(["qpid-config",
+ "--broker", src_cluster[0].host_port(),
+ "add", "exchange", "direct", "FedX"],
+ EXPECT_EXIT_OK)
+ cmd.wait()
+
+ cmd = self.popen(["qpid-config",
+ "--broker", dst_cluster[0].host_port(),
+ "add", "exchange", "direct", "FedX"],
+ EXPECT_EXIT_OK)
+ cmd.wait()
+
+ cmd = self.popen(["qpid-config",
+ "--broker", dst_cluster[0].host_port(),
+ "add", "queue", "destQ1"],
+ EXPECT_EXIT_OK)
+ cmd.wait()
+
+ cmd = self.popen(["qpid-config",
+ "--broker", dst_cluster[0].host_port(),
+ "bind", "FedX", "destQ1", "one"],
+ EXPECT_EXIT_OK)
+ cmd.wait()
+
+ cmd = self.popen(["qpid-config",
+ "--broker", dst_cluster[0].host_port(),
+ "add", "queue", "destQ2"],
+ EXPECT_EXIT_OK)
+ cmd.wait()
+
+ cmd = self.popen(["qpid-config",
+ "--broker", dst_cluster[0].host_port(),
+ "bind", "FedX", "destQ2", "two"],
+ EXPECT_EXIT_OK)
+ cmd.wait()
+
+ # Create two separate links between the dst and source brokers, bind
+ # each to different keys
+ dst_cluster[0].startQmf()
+ dst_broker = dst_cluster[0].qmf_session.getObjects(_class="broker")[0]
+
+ for _l in [("link1", "bridge1", "one"),
+ ("link2", "bridge2", "two")]:
+ result = dst_broker.create("link", _l[0],
+ {"host":src_cluster[0].host(),
+ "port":src_cluster[0].port()},
+ False)
+ self.assertEqual(result.status, 0, result);
+ result = dst_broker.create("bridge", _l[1],
+ {"link":_l[0],
+ "src":"FedX",
+ "dest":"FedX",
+ "key":_l[2]}, False)
+ self.assertEqual(result.status, 0);
+
+ # check that traffic passes
+ assert self._verify_federation(src_cluster[0], "FedX/one", dst_cluster[0], "destQ1")
+ assert self._verify_federation(src_cluster[0], "FedX/two", dst_cluster[0], "destQ2")
+
+ # add new member, verify traffic
+ src_cluster.start(expect=EXPECT_EXIT_FAIL);
+ src_cluster.ready();
+
+ dst_cluster.start(expect=EXPECT_EXIT_FAIL);
+ dst_cluster.ready();
+
+ assert self._verify_federation(src_cluster[0], "FedX/one", dst_cluster[0], "destQ1")
+ assert self._verify_federation(src_cluster[0], "FedX/two", dst_cluster[0], "destQ2")
+
+ src_cluster[0].kill()
+ for b in src_cluster[1:]: b.ready()
+
+ assert self._verify_federation(src_cluster[1], "FedX/one", dst_cluster[0], "destQ1")
+ assert self._verify_federation(src_cluster[1], "FedX/two", dst_cluster[0], "destQ2")
+
+ dst_cluster[0].kill()
+ for b in dst_cluster[1:]: b.ready()
+
+ assert self._verify_federation(src_cluster[1], "FedX/one", dst_cluster[1], "destQ1")
+ assert self._verify_federation(src_cluster[1], "FedX/two", dst_cluster[1], "destQ2")
+
+ for i in range(1, len(src_cluster)): src_cluster[i].kill()
+ for i in range(1, len(dst_cluster)): dst_cluster[i].kill()
+
+
+
# Some utility code for transaction tests
XA_RBROLLBACK = 1
XA_RBTIMEOUT = 2
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org