Author: gsim
Date: Fri May 24 16:49:28 2013
New Revision: 1486115
URL: http://svn.apache.org/r1486115
Log:
QPID-4888: correct handling of link naming
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedIncomingLink.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedOutgoingLink.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h
qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.h
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp
qpid/trunk/qpid/specs/management-schema.xml
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp Fri May 24 16:49:28
2013
@@ -179,6 +179,7 @@ void Connection::process()
if ((pn_connection_state(connection) & REQUIRES_OPEN) == REQUIRES_OPEN) {
QPID_LOG_CAT(debug, model, id << " connection opened");
pn_connection_set_container(connection,
broker.getFederationTag().c_str());
+ setContainerId(pn_connection_remote_container(connection));
pn_connection_open(connection);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.cpp Fri May 24
16:49:28 2013
@@ -22,6 +22,7 @@
#include "qpid/broker/Broker.h"
#include "qpid/management/ManagementAgent.h"
#include "qpid/log/Statement.h"
+#include "qpid/types/Variant.h"
#include "qmf/org/apache/qpid/broker/EventClientConnect.h"
#include "qmf/org/apache/qpid/broker/EventClientDisconnect.h"
@@ -77,6 +78,20 @@ void ManagedConnection::setSaslSsf(int s
}
}
+void ManagedConnection::setContainerId(const std::string& container)
+{
+ containerid = container;
+ if (connection) {
+ qpid::types::Variant::Map props;
+ props["container-id"] = containerid;
+ connection->set_remoteProperties(props);
+ }
+}
+const std::string& ManagedConnection::getContainerId() const
+{
+ return containerid;
+}
+
qpid::management::ManagementObject::shared_ptr
ManagedConnection::GetManagementObject() const
{
return connection;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.h?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedConnection.h Fri May 24
16:49:28 2013
@@ -44,6 +44,8 @@ class ManagedConnection : public qpid::m
std::string getUserid() const;
void setSaslMechanism(const std::string&);
void setSaslSsf(int);
+ void setContainerId(const std::string&);
+ const std::string& getContainerId() const;
qpid::management::ManagementObject::shared_ptr GetManagementObject() const;
bool isLocal(const ConnectionToken* t) const;
void incomingMessageReceived();
@@ -51,6 +53,7 @@ class ManagedConnection : public qpid::m
private:
const std::string id;
std::string userid;
+ std::string containerid;
qmf::org::apache::qpid::broker::Connection::shared_ptr connection;
qpid::management::ManagementAgent* agent;
};
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedIncomingLink.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedIncomingLink.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedIncomingLink.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedIncomingLink.cpp Fri May 24
16:49:28 2013
@@ -19,6 +19,7 @@
*
*/
#include "ManagedIncomingLink.h"
+#include "qpid/broker/amqp/ManagedConnection.h"
#include "qpid/broker/amqp/ManagedSession.h"
#include "qpid/broker/Broker.h"
#include "qpid/management/ManagementAgent.h"
@@ -35,7 +36,7 @@ ManagedIncomingLink::ManagedIncomingLink
{
qpid::management::ManagementAgent* agent = broker.getManagementAgent();
if (agent) {
- incoming = _qmf::Incoming::shared_ptr(new _qmf::Incoming(agent, this,
&parent, source, target, _name));
+ incoming = _qmf::Incoming::shared_ptr(new _qmf::Incoming(agent, this,
&parent, parent.getParent().getContainerId(), _name, source, target));
agent->addObject(incoming);
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedOutgoingLink.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedOutgoingLink.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedOutgoingLink.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedOutgoingLink.cpp Fri May 24
16:49:28 2013
@@ -19,6 +19,7 @@
*
*/
#include "ManagedOutgoingLink.h"
+#include "qpid/broker/amqp/ManagedConnection.h"
#include "qpid/broker/amqp/ManagedSession.h"
#include "qpid/broker/Broker.h"
#include "qpid/management/ManagementAgent.h"
@@ -35,7 +36,7 @@ ManagedOutgoingLink::ManagedOutgoingLink
{
qpid::management::ManagementAgent* agent = broker.getManagementAgent();
if (agent) {
- outgoing = _qmf::Outgoing::shared_ptr(new _qmf::Outgoing(agent, this,
&parent, source, target, _name));
+ outgoing = _qmf::Outgoing::shared_ptr(new _qmf::Outgoing(agent, this,
&parent, parent.getParent().getContainerId(), _name, source, target));
agent->addObject(outgoing);
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.cpp Fri May 24
16:49:28 2013
@@ -84,5 +84,9 @@ void ManagedSession::incomingMessageReje
{
}
+ManagedConnection& ManagedSession::getParent()
+{
+ return parent;
+}
}}} // namespace qpid::broker::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.h?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ManagedSession.h Fri May 24
16:49:28 2013
@@ -48,6 +48,7 @@ class ManagedSession : public qpid::mana
void outgoingMessageSent();
void outgoingMessageAccepted();
void outgoingMessageRejected();
+ ManagedConnection& getParent();
private:
ManagedConnection& parent;
const std::string id;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.cpp Fri May 24
16:49:28 2013
@@ -39,12 +39,13 @@ const std::string SUPPORTED_DIST_MODES("
//AMQP 0-10 standard parameters:
const std::string DURABLE("durable");
+const std::string EXCLUSIVE("exclusive");
const std::string AUTO_DELETE("auto-delete");
const std::string ALTERNATE_EXCHANGE("alternate-exchange");
const std::string EXCHANGE_TYPE("exchange-type");
}
-NodeProperties::NodeProperties() : queue(true), durable(false),
autoDelete(false), exchangeType("topic") {}
+NodeProperties::NodeProperties() : queue(true), durable(false),
autoDelete(false), exclusive(false), exchangeType("topic") {}
void NodeProperties::read(pn_data_t* data)
{
@@ -60,6 +61,8 @@ void NodeProperties::process(const std::
else if (value == COPY) queue = false;
} else if (key == DURABLE) {
durable = value;
+ } else if (key == EXCLUSIVE) {
+ exclusive = value;
} else if (key == AUTO_DELETE) {
autoDelete = value;
} else if (key == ALTERNATE_EXCHANGE) {
@@ -167,6 +170,10 @@ bool NodeProperties::isDurable() const
{
return durable;
}
+bool NodeProperties::isExclusive() const
+{
+ return exclusive;
+}
std::string NodeProperties::getExchangeType() const
{
return exchangeType;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/NodeProperties.h Fri May 24
16:49:28 2013
@@ -54,12 +54,14 @@ class NodeProperties : public qpid::amqp
bool isQueue() const;
QueueSettings getQueueSettings();
bool isDurable() const;
+ bool isExclusive() const;
std::string getExchangeType() const;
std::string getAlternateExchange() const;
private:
bool queue;
bool durable;
bool autoDelete;
+ bool exclusive;
std::string exchangeType;
std::string alternateExchange;
qpid::types::Variant::Map properties;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp Fri May 24 16:49:28
2013
@@ -36,7 +36,6 @@
#include "qpid/broker/Selector.h"
#include "qpid/broker/TopicExchange.h"
#include "qpid/broker/amqp/Filter.h"
-#include "qpid/broker/amqp/NodeProperties.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/framing/FieldTable.h"
#include "qpid/framing/MessageTransferBody.h"
@@ -133,13 +132,12 @@ Session::ResolvedNode Session::resolve(c
if (!node.queue && !node.exchange) {
if (pn_terminus_is_dynamic(terminus) ||
is_capability_requested(CREATE_ON_DEMAND, pn_terminus_capabilities(terminus))) {
//is it a queue or an exchange?
- NodeProperties properties;
- properties.read(pn_terminus_properties(terminus));
- if (properties.isQueue()) {
- node.queue = broker.createQueue(name,
properties.getQueueSettings(), this, properties.getAlternateExchange(),
connection.getUserid(), connection.getId()).first;
+ node.properties.read(pn_terminus_properties(terminus));
+ if (node.properties.isQueue()) {
+ node.queue = broker.createQueue(name,
node.properties.getQueueSettings(), this,
node.properties.getAlternateExchange(), connection.getUserid(),
connection.getId()).first;
} else {
qpid::framing::FieldTable args;
- node.exchange = broker.createExchange(name,
properties.getExchangeType(), properties.isDurable(),
properties.getAlternateExchange(),
+ node.exchange = broker.createExchange(name,
node.properties.getExchangeType(), node.properties.isDurable(),
node.properties.getAlternateExchange(),
args,
connection.getUserid(), connection.getId()).first;
}
} else {
@@ -164,6 +162,11 @@ Session::ResolvedNode Session::resolve(c
QPID_LOG_CAT(warning, protocol, "Ambiguous node name; " << name << "
could be queue or exchange, assuming queue");
node.exchange.reset();
}
+
+ if (node.properties.isExclusive() && node.queue &&
node.queue->setExclusiveOwner(this)) {
+ exclusiveQueues.insert(node.queue);
+ }
+
return node;
}
@@ -283,14 +286,17 @@ void Session::setupOutgoing(pn_link_t* l
}
outgoing[link] = q;
} else if (node.exchange) {
- QueueSettings settings(false, true);
+ bool durable = pn_terminus_get_durability(source);
+ QueueSettings settings(durable, !durable);
if (filter.hasSelectorFilter()) {
settings.filter = filter.getSelectorFilter();
QPID_LOG(debug, "Selector specified for outgoing link from
exchange " << node.exchange->getName() << ": " << settings.filter);
}
//TODO: populate settings from source details when available from
engine
+ std::stringstream queueName;//combination of container id and link
name is unique
+ queueName << connection.getContainerId() << "_" << pn_link_name(link);
boost::shared_ptr<qpid::broker::Queue> queue
- = broker.createQueue(name + qpid::types::Uuid(true).str(),
settings, this, "", connection.getUserid(), connection.getId()).first;
+ = broker.createQueue(queueName.str(), settings, this, "",
connection.getUserid(), connection.getId()).first;
queue->setExclusiveOwner(this);
if (filter.hasSubjectFilter()) {
filter.bind(node.exchange, queue);
@@ -430,6 +436,7 @@ bool Session::dispatch()
void Session::close()
{
+ exclusiveQueues.clear();
for (OutgoingLinks::iterator i = outgoing.begin(); i != outgoing.end();
++i) {
i->second->detached();
}
@@ -439,6 +446,9 @@ void Session::close()
outgoing.clear();
incoming.clear();
QPID_LOG(debug, "Session closed, all links detached.");
+ for (std::set< boost::shared_ptr<Queue> >::const_iterator i =
exclusiveQueues.begin(); i != exclusiveQueues.end(); ++i) {
+ (*i)->releaseExclusiveOwnership();
+ }
qpid::sys::Mutex::ScopedLock l(lock);
deleted = true;
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h Fri May 24 16:49:28 2013
@@ -24,8 +24,10 @@
#include "qpid/sys/Mutex.h"
#include "qpid/sys/OutputControl.h"
#include "qpid/broker/amqp/ManagedSession.h"
+#include "qpid/broker/amqp/NodeProperties.h"
#include <deque>
#include <map>
+#include <set>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
@@ -85,11 +87,13 @@ class Session : public ManagedSession, p
std::deque<pn_delivery_t*> completed;
bool deleted;
qpid::sys::Mutex lock;
+ std::set< boost::shared_ptr<Queue> > exclusiveQueues;
struct ResolvedNode
{
boost::shared_ptr<qpid::broker::Exchange> exchange;
boost::shared_ptr<qpid::broker::Queue> queue;
boost::shared_ptr<Relay> relay;
+ NodeProperties properties;
};
ResolvedNode resolve(const std::string name, pn_terminus_t* terminus, bool
incoming);
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.cpp Fri May 24
16:49:28 2013
@@ -113,6 +113,8 @@ void ConnectionOptions::set(const std::s
sslCertName = value.asString();
} else if (name == "x-reconnect-on-limit-exceeded" || name ==
"x_reconnect_on_limit_exceeded") {
reconnectOnLimitExceeded = value;
+ } else if (name == "container-id" || name == "container_id") {
+ identifier = value.asString();
} else {
throw qpid::messaging::MessagingException(QPID_MSG("Invalid option: "
<< name << " not recognised"));
}
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.h?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionOptions.h Fri May 24
16:49:28 2013
@@ -42,6 +42,7 @@ struct ConnectionOptions : qpid::client:
double maxReconnectInterval;
int32_t retries;
bool reconnectOnLimitExceeded;
+ std::string identifier;
ConnectionOptions(const std::map<std::string, qpid::types::Variant>&);
void set(const std::string& name, const qpid::types::Variant& value);
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp Fri May 24
16:49:28 2013
@@ -24,7 +24,9 @@
#include "qpid/log/Statement.h"
#include <vector>
#include <set>
+#include <sstream>
#include <boost/assign.hpp>
+#include <boost/format.hpp>
extern "C" {
#include <proton/engine.h>
}
@@ -59,6 +61,9 @@ const std::string TYPE("type");
const std::string TOPIC("topic");
const std::string QUEUE("queue");
const std::string DURABLE("durable");
+const std::string NAME("name");
+const std::string RELIABILITY("reliability");
+const std::string SELECTOR("selector");
//distribution modes:
const std::string MOVE("move");
@@ -77,6 +82,17 @@ const std::string ARGUMENTS("arguments")
const std::vector<std::string> RECEIVER_MODES =
boost::assign::list_of<std::string>(ALWAYS) (RECEIVER);
const std::vector<std::string> SENDER_MODES =
boost::assign::list_of<std::string>(ALWAYS) (SENDER);
+class Verifier
+{
+ public:
+ Verifier();
+ void verify(const Address& address) const;
+ private:
+ Variant::Map defined;
+ void verify(const Variant::Map& allowed, const Variant::Map& actual) const;
+};
+const Verifier verifier;
+
pn_bytes_t convert(const std::string& s)
{
pn_bytes_t result;
@@ -177,6 +193,7 @@ AddressHelper::AddressHelper(const Addre
durableLink(false),
browse(false)
{
+ verifier.verify(address);
bind(address, CREATE, createPolicy);
bind(address, DELETE, deletePolicy);
bind(address, ASSERT, assertPolicy);
@@ -191,6 +208,7 @@ AddressHelper::AddressHelper(const Addre
if (bind(address, MODE, mode)) {
if (mode == BROWSE) {
browse = true;
+ throw qpid::messaging::AddressError("Browse mode not yet supported
over AMQP 1.0.");
} else if (mode != CONSUME) {
throw qpid::messaging::AddressError("Invalid value for mode; must
be 'browse' or 'consume'.");
}
@@ -325,6 +343,19 @@ void AddressHelper::setCapabilities(pn_t
pn_data_put_symbol(data, convert(i->asString()));
}
}
+std::string AddressHelper::getLinkName(const Address& address)
+{
+ AddressHelper helper(address);
+ const qpid::types::Variant::Map& linkProps = helper.getLinkProperties();
+ qpid::types::Variant::Map::const_iterator i = linkProps.find(NAME);
+ if (i != linkProps.end()) {
+ return i->second.asString();
+ } else {
+ std::stringstream name;
+ name << address.getName() << "_" << qpid::types::Uuid(true);
+ return name.str();
+ }
+}
void AddressHelper::setNodeProperties(pn_terminus_t* terminus)
{
@@ -348,4 +379,45 @@ void AddressHelper::setNodeProperties(pn
}
}
+Verifier::Verifier()
+{
+ defined[CREATE] = true;
+ defined[ASSERT] = true;
+ defined[DELETE] = true;
+ defined[MODE] = true;
+ Variant::Map node;
+ node[TYPE] = true;
+ node[DURABLE] = true;
+ node[PROPERTIES] = true;
+ node[CAPABILITIES] = true;
+ node[X_DECLARE] = true;
+ node[X_BINDINGS] = true;
+ defined[NODE] = node;
+ Variant::Map link;
+ link[NAME] = true;
+ link[DURABLE] = true;
+ link[RELIABILITY] = true;
+ link[X_SUBSCRIBE] = true;
+ link[X_DECLARE] = true;
+ link[X_BINDINGS] = true;
+ link[SELECTOR] = true;
+ defined[LINK] = link;
+}
+void Verifier::verify(const Address& address) const
+{
+ verify(defined, address.getOptions());
+}
+
+void Verifier::verify(const Variant::Map& allowed, const Variant::Map& actual)
const
+{
+ for (Variant::Map::const_iterator i = actual.begin(); i != actual.end();
++i) {
+ Variant::Map::const_iterator option = allowed.find(i->first);
+ if (option == allowed.end()) {
+ throw AddressError((boost::format("Unrecognised option: %1%") %
i->first).str());
+ } else if (option->second.getType() == qpid::types::VAR_MAP) {
+ verify(option->second.asMap(), i->second.asMap());
+ }
+ }
+}
+
}}} // namespace qpid::messaging::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h Fri May 24
16:49:28 2013
@@ -41,6 +41,7 @@ class AddressHelper
const qpid::types::Variant::Map& getNodeProperties() const;
const qpid::types::Variant::Map& getLinkProperties() const;
+ static std::string getLinkName(const Address& address);
private:
bool isTemporary;
std::string createPolicy;
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Fri May
24 16:49:28 2013
@@ -60,7 +60,10 @@ ConnectionContext::ConnectionContext(con
if (pn_transport_bind(engine, connection)) {
//error
}
- pn_connection_set_container(connection, "qpid::messaging");//TODO: take
this from a connection option
+ if (identifier.empty()) {
+ identifier = qpid::types::Uuid(true).str();
+ }
+ pn_connection_set_container(connection, identifier.c_str());
bool enableTrace(false);
QPID_LOG_TEST_CAT(trace, protocol, enableTrace);
if (enableTrace) pn_transport_trace(engine, PN_TRACE_FRM);
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp Fri May 24
16:49:28 2013
@@ -43,12 +43,8 @@ SessionContext::~SessionContext()
boost::shared_ptr<SenderContext> SessionContext::createSender(const
qpid::messaging::Address& address)
{
- std::string name = address.getName();
-
- int count = 1;
- for (SenderMap::const_iterator i = senders.find(name); i != senders.end();
i = senders.find(name)) {
- name = (boost::format("%1%_%2%") % address.getName() % ++count).str();
- }
+ std::string name = AddressHelper::getLinkName(address);
+ if (senders.find(name) != senders.end()) throw LinkError("Link name must
be unique within the scope of the connection");
boost::shared_ptr<SenderContext> s(new SenderContext(session, name,
address));
senders[name] = s;
return s;
@@ -56,12 +52,8 @@ boost::shared_ptr<SenderContext> Session
boost::shared_ptr<ReceiverContext> SessionContext::createReceiver(const
qpid::messaging::Address& address)
{
- std::string name = address.getName();
-
- int count = 1;
- for (ReceiverMap::const_iterator i = receivers.find(name); i !=
receivers.end(); i = receivers.find(name)) {
- name = (boost::format("%1%_%2%") % address.getName() % ++count).str();
- }
+ std::string name = AddressHelper::getLinkName(address);
+ if (receivers.find(name) != receivers.end()) throw LinkError("Link name
must be unique within the scope of the connection");
boost::shared_ptr<ReceiverContext> r(new ReceiverContext(session, name,
address));
receivers[name] = r;
return r;
Modified: qpid/trunk/qpid/specs/management-schema.xml
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/specs/management-schema.xml?rev=1486115&r1=1486114&r2=1486115&view=diff
==============================================================================
--- qpid/trunk/qpid/specs/management-schema.xml (original)
+++ qpid/trunk/qpid/specs/management-schema.xml Fri May 24 16:49:28 2013
@@ -394,9 +394,10 @@
-->
<class name="Incoming">
<property name="sessionRef" type="objId" references="Session"
access="RC" parentRef="y"/>
- <property name="source" type="sstr" access="RC" index="y"/>
- <property name="target" type="sstr" access="RC" index="y"/>
+ <property name="containerid" type="sstr" access="RC" index="y"/>
<property name="name" type="sstr" access="RC" index="y"/>
+ <property name="source" type="sstr" access="RC"/>
+ <property name="target" type="sstr" access="RC"/>
<statistic name="transfers" type="count64" unit="message"
desc="Messages transfered"/>
</class>
<!--
@@ -406,9 +407,10 @@
-->
<class name="Outgoing">
<property name="sessionRef" type="objId" references="Session"
access="RC" parentRef="y"/>
- <property name="source" type="sstr" access="RC" index="y"/>
- <property name="target" type="sstr" access="RC" index="y"/>
+ <property name="containerid" type="sstr" access="RC" index="y"/>
<property name="name" type="sstr" access="RC" index="y"/>
+ <property name="source" type="sstr" access="RC"/>
+ <property name="target" type="sstr" access="RC"/>
<statistic name="transfers" type="count64" unit="message"
desc="Messages transfered"/>
</class>
<!--
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
