Author: gsim
Date: Sat Mar 23 17:59:50 2013
New Revision: 1460198
URL: http://svn.apache.org/r1460198
Log:
QPID-4586: fixes for dynamic sources/targets and on demand creation of nodes
Added:
qpid/trunk/qpid/cpp/src/qpid/messaging/AddressImpl.h
- copied, changed from r1460013,
qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h
Modified:
qpid/trunk/qpid/cpp/examples/messaging/client.cpp
qpid/trunk/qpid/cpp/include/qpid/messaging/Address.h
qpid/trunk/qpid/cpp/include/qpid/messaging/Receiver.h
qpid/trunk/qpid/cpp/include/qpid/messaging/Sender.h
qpid/trunk/qpid/cpp/src/CMakeLists.txt
qpid/trunk/qpid/cpp/src/Makefile.am
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Interconnect.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.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/client/amqp0_10/ReceiverImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
qpid/trunk/qpid/cpp/src/qpid/messaging/Address.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/AddressParser.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/Receiver.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/ReceiverImpl.h
qpid/trunk/qpid/cpp/src/qpid/messaging/Sender.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.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/ReceiverContext.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h
Modified: qpid/trunk/qpid/cpp/examples/messaging/client.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/examples/messaging/client.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/examples/messaging/client.cpp (original)
+++ qpid/trunk/qpid/cpp/examples/messaging/client.cpp Sat Mar 23 17:59:50 2013
@@ -48,8 +48,8 @@ int main(int argc, char** argv) {
Sender sender = session.createSender("service_queue");
//create temp queue & receiver...
- Address responseQueue("#response-queue; {create:always,
delete:always}");
- Receiver receiver = session.createReceiver(responseQueue);
+ Receiver receiver = session.createReceiver("#");
+ Address responseQueue = receiver.getAddress();
// Now send some messages ...
string s[] = {
Modified: qpid/trunk/qpid/cpp/include/qpid/messaging/Address.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/include/qpid/messaging/Address.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/include/qpid/messaging/Address.h (original)
+++ qpid/trunk/qpid/cpp/include/qpid/messaging/Address.h Sat Mar 23 17:59:50
2013
@@ -153,6 +153,7 @@ class QPID_MESSAGING_CLASS_EXTERN Addres
QPID_MESSAGING_EXTERN bool operator !() const;
private:
AddressImpl* impl;
+ friend class AddressImpl;
};
#ifndef SWIG
Modified: qpid/trunk/qpid/cpp/include/qpid/messaging/Receiver.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/include/qpid/messaging/Receiver.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/include/qpid/messaging/Receiver.h (original)
+++ qpid/trunk/qpid/cpp/include/qpid/messaging/Receiver.h Sat Mar 23 17:59:50
2013
@@ -34,6 +34,7 @@ namespace messaging {
template <class> class PrivateImplRef;
#endif
+class Address;
class Message;
class ReceiverImpl;
class Session;
@@ -134,6 +135,11 @@ class QPID_MESSAGING_CLASS_EXTERN Receiv
*/
QPID_MESSAGING_EXTERN Session getSession() const;
+ /**
+ * Returns an address for this receiver.
+ */
+ QPID_MESSAGING_EXTERN Address getAddress() const;
+
#ifndef SWIG
private:
friend class qpid::messaging::PrivateImplRef<Receiver>;
Modified: qpid/trunk/qpid/cpp/include/qpid/messaging/Sender.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/include/qpid/messaging/Sender.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/include/qpid/messaging/Sender.h (original)
+++ qpid/trunk/qpid/cpp/include/qpid/messaging/Sender.h Sat Mar 23 17:59:50 2013
@@ -34,6 +34,7 @@ namespace messaging {
#ifndef SWIG
template <class> class PrivateImplRef;
#endif
+class Address;
class Message;
class SenderImpl;
class Session;
@@ -89,6 +90,11 @@ class QPID_MESSAGING_CLASS_EXTERN Sender
* Returns a handle to the session associated with this sender.
*/
QPID_MESSAGING_EXTERN Session getSession() const;
+
+ /**
+ * Returns an address for this sender.
+ */
+ QPID_MESSAGING_EXTERN Address getAddress() const;
#ifndef SWIG
private:
friend class qpid::messaging::PrivateImplRef<Sender>;
Modified: qpid/trunk/qpid/cpp/src/CMakeLists.txt
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/CMakeLists.txt?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/CMakeLists.txt (original)
+++ qpid/trunk/qpid/cpp/src/CMakeLists.txt Sat Mar 23 17:59:50 2013
@@ -1063,6 +1063,7 @@ install_pdb (qpidclient ${QPID_COMPONENT
set (qpidmessaging_SOURCES_hidden
qpid/messaging/AddressParser.h
+ qpid/messaging/AddressImpl.h
qpid/messaging/ConnectionImpl.h
qpid/messaging/ReceiverImpl.h
qpid/messaging/SessionImpl.h
Modified: qpid/trunk/qpid/cpp/src/Makefile.am
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/Makefile.am?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/Makefile.am Sat Mar 23 17:59:50 2013
@@ -935,6 +935,7 @@ libqpidtypes_la_LDFLAGS = -version-info
libqpidmessaging_la_LIBADD = libqpidclient.la libqpidtypes.la
libqpidmessaging_la_SOURCES = \
qpid/messaging/Address.cpp \
+ qpid/messaging/AddressImpl.h \
qpid/messaging/AddressParser.h \
qpid/messaging/AddressParser.cpp \
qpid/messaging/Connection.cpp \
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=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.cpp Sat Mar 23 17:59:50
2013
@@ -37,11 +37,11 @@ namespace qpid {
namespace broker {
namespace amqp {
-Connection::Connection(qpid::sys::OutputControl& o, const std::string& i,
qpid::broker::Broker& b, Interconnects& interconnects_, bool saslInUse)
+Connection::Connection(qpid::sys::OutputControl& o, const std::string& i,
qpid::broker::Broker& b, Interconnects& interconnects_, bool saslInUse, const
std::string& d)
: ManagedConnection(b, i),
connection(pn_connection()),
transport(pn_transport()),
- out(o), id(i), broker(b), haveOutput(true), interconnects(interconnects_)
+ out(o), id(i), broker(b), haveOutput(true),
interconnects(interconnects_), domain(d)
{
if (pn_transport_bind(transport, connection)) {
//error
@@ -265,4 +265,8 @@ std::string Connection::getError()
return text.str();
}
+std::string Connection::getDomain() const
+{
+ return domain;
+}
}}} // namespace qpid::broker::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Connection.h Sat Mar 23 17:59:50
2013
@@ -45,7 +45,7 @@ class Session;
class Connection : public sys::ConnectionCodec, public ManagedConnection
{
public:
- Connection(qpid::sys::OutputControl& out, const std::string& id,
qpid::broker::Broker& broker, Interconnects&, bool saslInUse);
+ Connection(qpid::sys::OutputControl& out, const std::string& id,
qpid::broker::Broker& broker, Interconnects&, bool saslInUse, const
std::string& domain);
virtual ~Connection();
size_t decode(const char* buffer, size_t size);
virtual size_t encode(char* buffer, size_t size);
@@ -57,6 +57,7 @@ class Connection : public sys::Connectio
framing::ProtocolVersion getVersion() const;
pn_transport_t* getTransport();
Interconnects& getInterconnects();
+ std::string getDomain() const;
protected:
typedef std::map<pn_session_t*, boost::shared_ptr<Session> > Sessions;
pn_connection_t* connection;
@@ -67,6 +68,7 @@ class Connection : public sys::Connectio
bool haveOutput;
Sessions sessions;
Interconnects& interconnects;
+ std::string domain;
virtual void process();
std::string getError();
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Interconnect.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Interconnect.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Interconnect.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Interconnect.cpp Sat Mar 23
17:59:50 2013
@@ -41,7 +41,7 @@ namespace amqp {
Interconnect::Interconnect(qpid::sys::OutputControl& out, const std::string&
id, qpid::broker::Broker& broker, bool saslInUse,
bool i, const std::string& n, const std::string& s,
const std::string& t, Domain& d, Interconnects& r)
- : Connection(out, id, broker, r, saslInUse), incoming(i), name(n),
source(s), target(t), domain(d), registry(r), headerDiscarded(false),
+ : Connection(out, id, broker, r, saslInUse, std::string()), incoming(i),
name(n), source(s), target(t), domain(d), registry(r), headerDiscarded(false),
closeRequested(false), isTransportDeleted(false)
{}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp Sat Mar 23
17:59:50 2013
@@ -41,10 +41,19 @@ namespace qpid {
namespace broker {
namespace amqp {
+struct Options : public qpid::Options {
+ std::string domain;
+
+ Options() : qpid::Options("AMQP 1.0 Options") {
+ addOptions()
+ ("domain", optValue(domain, "DOMAIN"), "Domain of this broker");
+ }
+};
+
class ProtocolImpl : public Protocol
{
public:
- ProtocolImpl(Interconnects* i, Broker& b) : interconnects(i), broker(b)
+ ProtocolImpl(Interconnects* i, Broker& b, const std::string& d) :
interconnects(i), broker(b), domain(d)
{
broker.getObjectFactoryRegistry().add(interconnects);//registry
deletes on shutdown
}
@@ -54,16 +63,20 @@ class ProtocolImpl : public Protocol
private:
Interconnects* interconnects;
Broker& broker;
+ std::string domain;
};
struct ProtocolPlugin : public Plugin
{
+ Options options;
+ Options* getOptions() { return &options; }
+
void earlyInitialize(Plugin::Target& target)
{
//need to register protocol before recovery from store
broker::Broker* broker = dynamic_cast<qpid::broker::Broker*>(&target);
if (broker) {
- ProtocolImpl* impl = new ProtocolImpl(new Interconnects(),
*broker);
+ ProtocolImpl* impl = new ProtocolImpl(new Interconnects(),
*broker, options.domain);
broker->getProtocolRegistry().add("AMQP 1.0", impl);//registry
deletes on shutdown
}
}
@@ -79,18 +92,20 @@ qpid::sys::ConnectionCodec* ProtocolImpl
if (v.getProtocol() == qpid::framing::ProtocolVersion::SASL) {
if (broker.getOptions().auth) {
QPID_LOG(info, "Using AMQP 1.0 (with SASL layer)");
- return new qpid::broker::amqp::Sasl(out, id, broker,
*interconnects,
qpid::SaslFactory::getInstance().createServer(broker.getOptions().realm,
broker.getOptions().requireEncrypted, external));
+ return new qpid::broker::amqp::Sasl(out, id, broker,
*interconnects,
+
qpid::SaslFactory::getInstance().createServer(broker.getOptions().realm,broker.getOptions().requireEncrypted,
external),
+ domain);
} else {
std::auto_ptr<SaslServer> authenticator(new
qpid::NullSaslServer(broker.getOptions().realm));
QPID_LOG(info, "Using AMQP 1.0 (with dummy SASL layer)");
- return new qpid::broker::amqp::Sasl(out, id, broker,
*interconnects, authenticator);
+ return new qpid::broker::amqp::Sasl(out, id, broker,
*interconnects, authenticator, domain);
}
} else {
if (broker.getOptions().auth) {
throw qpid::Exception("SASL layer required!");
} else {
QPID_LOG(info, "Using AMQP 1.0 (no SASL layer)");
- return new qpid::broker::amqp::Connection(out, id, broker,
*interconnects, false);
+ return new qpid::broker::amqp::Connection(out, id, broker,
*interconnects, false, domain);
}
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.cpp Sat Mar 23 17:59:50 2013
@@ -31,8 +31,8 @@ namespace qpid {
namespace broker {
namespace amqp {
-Sasl::Sasl(qpid::sys::OutputControl& o, const std::string& id,
qpid::broker::Broker& broker, Interconnects& i, std::auto_ptr<qpid::SaslServer>
auth)
- : qpid::amqp::SaslServer(id), out(o), connection(out, id, broker, i, true),
+Sasl::Sasl(qpid::sys::OutputControl& o, const std::string& id,
qpid::broker::Broker& broker, Interconnects& i, std::auto_ptr<qpid::SaslServer>
auth, const std::string& domain)
+ : qpid::amqp::SaslServer(id), out(o), connection(out, id, broker, i, true,
domain),
authenticator(auth),
state(INCOMPLETE), writeHeader(true), haveOutput(true)
{
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Sasl.h Sat Mar 23 17:59:50 2013
@@ -39,7 +39,7 @@ namespace amqp {
class Sasl : public sys::ConnectionCodec, qpid::amqp::SaslServer
{
public:
- Sasl(qpid::sys::OutputControl& out, const std::string& id,
qpid::broker::Broker& broker, Interconnects&, std::auto_ptr<qpid::SaslServer>
authenticator);
+ Sasl(qpid::sys::OutputControl& out, const std::string& id,
qpid::broker::Broker& broker, Interconnects&, std::auto_ptr<qpid::SaslServer>
authenticator, const std::string& domain);
~Sasl();
size_t decode(const char* buffer, size_t size);
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=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp Sat Mar 23 17:59:50
2013
@@ -53,6 +53,20 @@ namespace qpid {
namespace broker {
namespace amqp {
+namespace {
+bool is_capability_requested(const std::string& name, pn_data_t* capabilities)
+{
+ while (pn_data_next(capabilities)) {
+ pn_bytes_t c = pn_data_get_symbol(capabilities);
+ std::string s(c.start, c.size);
+ if (s == name) return true;
+ }
+ return false;
+}
+
+const std::string CREATE_ON_DEMAND("create-on-demand");
+}
+
class IncomingToQueue : public DecodingIncoming
{
public:
@@ -81,7 +95,7 @@ Session::ResolvedNode Session::resolve(c
node.exchange = broker.getExchanges().find(name);
node.queue = broker.getQueues().find(name);
if (!node.queue && !node.exchange) {
- if (pn_terminus_is_dynamic(terminus)) {
+ 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));
@@ -117,25 +131,43 @@ Session::ResolvedNode Session::resolve(c
return node;
}
+std::string Session::generateName(pn_link_t* link)
+{
+ std::stringstream s;
+ s << qpid::types::Uuid(true) << "::" << pn_link_name(link);
+ if (!connection.getDomain().empty()) {
+ s << "@" << connection.getDomain();
+ }
+ return s.str();
+}
+
void Session::attach(pn_link_t* link)
{
if (pn_link_is_sender(link)) {
pn_terminus_t* source = pn_link_remote_source(link);
//i.e a subscription
+ std::string name;
if (pn_terminus_get_type(source) == PN_UNSPECIFIED) {
throw qpid::Exception("No source specified!");/*invalid-field?*/
+ } else if (pn_terminus_is_dynamic(source)) {
+ name = generateName(link);
+ } else {
+ name = pn_terminus_get_address(source);
}
- std::string name = pn_terminus_get_address(source);
QPID_LOG(debug, "Received attach request for outgoing link from " <<
name);
pn_terminus_set_address(pn_link_source(link), name.c_str());
setupOutgoing(link, source, name);
} else {
pn_terminus_t* target = pn_link_remote_target(link);
+ std::string name;
if (pn_terminus_get_type(target) == PN_UNSPECIFIED) {
throw qpid::Exception("No target specified!");/*invalid field?*/
+ } else if (pn_terminus_is_dynamic(target)) {
+ name = generateName(link);
+ } else {
+ name = pn_terminus_get_address(target);
}
- std::string name = pn_terminus_get_address(target);
QPID_LOG(debug, "Received attach request for incoming link to " <<
name);
pn_terminus_set_address(pn_link_target(link), name.c_str());
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=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.h Sat Mar 23 17:59:50 2013
@@ -95,6 +95,7 @@ class Session : public ManagedSession, p
ResolvedNode resolve(const std::string name, pn_terminus_t* terminus, bool
incoming);
void setupOutgoing(pn_link_t* link, pn_terminus_t* source, const
std::string& name);
void setupIncoming(pn_link_t* link, pn_terminus_t* target, const
std::string& name);
+ std::string generateName(pn_link_t*);
};
}}} // namespace qpid::broker::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp Sat Mar 23
17:59:50 2013
@@ -143,6 +143,11 @@ uint32_t ReceiverImpl::getUnsettled()
return parent->getUnsettledAcks(destination);
}
+qpid::messaging::Address ReceiverImpl::getAddress() const
+{
+ return address;
+}
+
ReceiverImpl::ReceiverImpl(SessionImpl& p, const std::string& name,
const qpid::messaging::Address& a) :
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.h Sat Mar 23
17:59:50 2013
@@ -66,6 +66,7 @@ class ReceiverImpl : public qpid::messag
void received(qpid::messaging::Message& message);
qpid::messaging::Session getSession() const;
bool isClosed() const;
+ qpid::messaging::Address getAddress() const;
private:
mutable sys::Mutex lock;
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp Sat Mar 23
17:59:50 2013
@@ -34,6 +34,11 @@ SenderImpl::SenderImpl(SessionImpl& _par
parent(&_parent), name(_name), address(_address), state(UNRESOLVED),
capacity(50), window(0), flushed(false),
unreliable(AddressResolution::is_unreliable(address)) {}
+qpid::messaging::Address SenderImpl::getAddress() const
+{
+ return address;
+}
+
void SenderImpl::send(const qpid::messaging::Message& message, bool sync)
{
if (unreliable) { // immutable, don't need lock
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h Sat Mar 23
17:59:50 2013
@@ -56,6 +56,7 @@ class SenderImpl : public qpid::messagin
void init(qpid::client::AsyncSession, AddressResolution&);
const std::string& getName() const;
qpid::messaging::Session getSession() const;
+ qpid::messaging::Address getAddress() const;
private:
mutable sys::Mutex lock;
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/Address.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/Address.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/Address.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/Address.cpp Sat Mar 23 17:59:50 2013
@@ -19,6 +19,8 @@
*
*/
#include "qpid/messaging/Address.h"
+#include "qpid/messaging/AddressImpl.h"
+#include "qpid/messaging/AddressParser.h"
#include "qpid/framing/Uuid.h"
#include <sstream>
#include <boost/format.hpp>
@@ -34,51 +36,9 @@ const std::string OPTIONS_DIVIDER = ";";
const std::string SPACE = " ";
const std::string TYPE = "type";
}
-class AddressImpl
-{
- public:
- std::string name;
- std::string subject;
- Variant::Map options;
-
- AddressImpl() {}
- AddressImpl(const std::string& n, const std::string& s, const
Variant::Map& o) :
- name(n), subject(s), options(o) {}
-};
-
-class AddressParser
-{
- public:
- AddressParser(const std::string&);
- bool parse(Address& address);
- private:
- const std::string& input;
- std::string::size_type current;
- static const std::string RESERVED;
-
- bool readChar(char c);
- bool readQuotedString(std::string& s);
- bool readQuotedValue(Variant& value);
- bool readString(std::string& value, char delimiter);
- bool readWord(std::string& word, const std::string& delims = RESERVED);
- bool readSimpleValue(Variant& word);
- bool readKey(std::string& key);
- bool readValue(Variant& value);
- bool readKeyValuePair(Variant::Map& map);
- bool readMap(Variant& value);
- bool readList(Variant& value);
- bool readName(std::string& name);
- bool readSubject(std::string& subject);
- bool error(const std::string& message);
- bool eos();
- bool iswhitespace();
- bool in(const std::string& delims);
- bool isreserved();
-};
-
Address::Address() : impl(new AddressImpl()) {}
Address::Address(const std::string& address) : impl(new AddressImpl())
-{
+{
AddressParser parser(address);
parser.parse(*this);
}
@@ -86,7 +46,7 @@ Address::Address(const std::string& name
const std::string& type)
: impl(new AddressImpl(name, subject, options)) { setType(type); }
Address::Address(const Address& a) :
- impl(new AddressImpl(a.impl->name, a.impl->subject, a.impl->options)) {}
+ impl(new AddressImpl(a.impl->name, a.impl->subject, a.impl->options)) {
impl->temporary = a.impl->temporary; }
Address::~Address() { delete impl; }
Address& Address::operator=(const Address& a) { *impl = *a.impl; return *this;
}
Copied: qpid/trunk/qpid/cpp/src/qpid/messaging/AddressImpl.h (from r1460013,
qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h)
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/AddressImpl.h?p2=qpid/trunk/qpid/cpp/src/qpid/messaging/AddressImpl.h&p1=qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h&r1=1460013&r2=1460198&rev=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/AddressImpl.h Sat Mar 23 17:59:50
2013
@@ -1,5 +1,5 @@
-#ifndef QPID_MESSAGING_SENDERIMPL_H
-#define QPID_MESSAGING_SENDERIMPL_H
+#ifndef QPID_MESSAGING_ADDRESSIMPL_H
+#define QPID_MESSAGING_ADDRESSIMPL_H
/*
*
@@ -10,9 +10,9 @@
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -21,28 +21,24 @@
* under the License.
*
*/
-#include "qpid/RefCounted.h"
-#include "qpid/sys/IntegerTypes.h"
-
+#include "qpid/types/Variant.h"
namespace qpid {
namespace messaging {
-class Message;
-class Session;
-
-class SenderImpl : public virtual qpid::RefCounted
+class AddressImpl
{
public:
- virtual ~SenderImpl() {}
- virtual void send(const Message& message, bool sync) = 0;
- virtual void close() = 0;
- virtual void setCapacity(uint32_t) = 0;
- virtual uint32_t getCapacity() = 0;
- virtual uint32_t getUnsettled() = 0;
- virtual const std::string& getName() const = 0;
- virtual Session getSession() const = 0;
- private:
+ std::string name;
+ std::string subject;
+ qpid::types::Variant::Map options;
+ bool temporary;
+
+ AddressImpl() : temporary(false) {}
+ AddressImpl(const std::string& n, const std::string& s, const
qpid::types::Variant::Map& o) :
+ name(n), subject(s), options(o), temporary(false) {}
+ static void setTemporary(Address& a, bool value) { a.impl->temporary =
value; }
+ static bool isTemporary(const Address& a) { return a.impl->temporary; }
};
}} // namespace qpid::messaging
-#endif /*!QPID_MESSAGING_SENDERIMPL_H*/
+#endif /*!QPID_MESSAGING_ADDRESSIMPL_H*/
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/AddressParser.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/AddressParser.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/AddressParser.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/AddressParser.cpp Sat Mar 23
17:59:50 2013
@@ -19,6 +19,7 @@
*
*/
#include "AddressParser.h"
+#include "AddressImpl.h"
#include "qpid/framing/Uuid.h"
#include <boost/format.hpp>
@@ -38,7 +39,10 @@ bool AddressParser::parse(Address& addre
{
std::string name;
if (readName(name)) {
- if (name.find('#') == 0) name = qpid::framing::Uuid(true).str() + name;
+ if (name.find('#') == 0) {
+ name = qpid::framing::Uuid(true).str() + name;
+ AddressImpl::setTemporary(address, true);
+ }
address.setName(name);
if (readChar('/')) {
std::string subject;
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/Receiver.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/Receiver.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/Receiver.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/Receiver.cpp Sat Mar 23 17:59:50 2013
@@ -19,6 +19,7 @@
*
*/
#include "qpid/messaging/Receiver.h"
+#include "qpid/messaging/Address.h"
#include "qpid/messaging/Message.h"
#include "qpid/messaging/ReceiverImpl.h"
#include "qpid/messaging/Session.h"
@@ -45,4 +46,5 @@ void Receiver::close() { impl->close();
const std::string& Receiver::getName() const { return impl->getName(); }
Session Receiver::getSession() const { return impl->getSession(); }
bool Receiver::isClosed() const { return impl->isClosed(); }
+Address Receiver::getAddress() const { return impl->getAddress(); }
}} // namespace qpid::messaging
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/ReceiverImpl.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/ReceiverImpl.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/ReceiverImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/ReceiverImpl.h Sat Mar 23 17:59:50
2013
@@ -27,6 +27,7 @@
namespace qpid {
namespace messaging {
+class Address;
class Duration;
class Message;
class MessageListener;
@@ -48,6 +49,7 @@ class ReceiverImpl : public virtual qpid
virtual const std::string& getName() const = 0;
virtual Session getSession() const = 0;
virtual bool isClosed() const = 0;
+ virtual Address getAddress() const = 0;
};
}} // namespace qpid::messaging
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/Sender.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/Sender.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/Sender.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/Sender.cpp Sat Mar 23 17:59:50 2013
@@ -19,6 +19,7 @@
*
*/
#include "qpid/messaging/Sender.h"
+#include "qpid/messaging/Address.h"
#include "qpid/messaging/Message.h"
#include "qpid/messaging/SenderImpl.h"
#include "qpid/messaging/Session.h"
@@ -40,5 +41,5 @@ uint32_t Sender::getUnsettled() { return
uint32_t Sender::getAvailable() { return getCapacity() - getUnsettled(); }
const std::string& Sender::getName() const { return impl->getName(); }
Session Sender::getSession() const { return impl->getSession(); }
-
+Address Sender::getAddress() const { return impl->getAddress(); }
}} // namespace qpid::messaging
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h Sat Mar 23 17:59:50 2013
@@ -27,6 +27,7 @@
namespace qpid {
namespace messaging {
+class Address;
class Message;
class Session;
@@ -41,6 +42,7 @@ class SenderImpl : public virtual qpid::
virtual uint32_t getUnsettled() = 0;
virtual const std::string& getName() const = 0;
virtual Session getSession() const = 0;
+ virtual Address getAddress() const = 0;
private:
};
}} // namespace qpid::messaging
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=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp Sat Mar 23
17:59:50 2013
@@ -20,6 +20,7 @@
*/
#include "qpid/messaging/amqp/AddressHelper.h"
#include "qpid/messaging/Address.h"
+#include "qpid/log/Statement.h"
#include <vector>
#include <boost/assign.hpp>
extern "C" {
@@ -56,7 +57,9 @@ const std::string MOVE("move");
const std::string COPY("copy");
const std::string SUPPORTED_DIST_MODES("supported-dist-modes");
+const std::string CREATE_ON_DEMAND("create-on-demand");
+const std::string DUMMY(".");
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);
@@ -155,26 +158,33 @@ const qpid::types::Variant::Map& Address
return link;
}
-void AddressHelper::setNodeProperties(pn_terminus_t* terminus)
+void AddressHelper::setNodeProperties(pn_terminus_t* terminus, bool dynamic)
{
- pn_terminus_set_dynamic(terminus, true);
+ if (dynamic) {
+ pn_terminus_set_address(terminus, DUMMY.c_str());//Workaround for
proton bug
+ pn_terminus_set_dynamic(terminus, true);
+ } else {
+ pn_data_t* capabilities = pn_terminus_capabilities(terminus);
+ if (!capabilities) {
+ QPID_LOG(error, "!!!No capabilities!!!");
+ }
+ pn_data_put_symbol(capabilities, convert(CREATE_ON_DEMAND));
+ }
//properties for dynamically created node:
- pn_data_t* data = pn_terminus_properties(terminus);
if (node.size()) {
+ pn_data_t* data = pn_terminus_properties(terminus);
pn_data_put_map(data);
pn_data_enter(data);
- }
- for (qpid::types::Variant::Map::const_iterator i = node.begin(); i !=
node.end(); ++i) {
- if (i->first == TYPE) {
- pn_data_put_symbol(data, convert(SUPPORTED_DIST_MODES));
- pn_data_put_string(data, convert(i->second == TOPIC ? COPY :
MOVE));
- } else {
- pn_data_put_symbol(data, convert(i->first));
- pn_data_put_string(data, convert(i->second.asString()));
+ for (qpid::types::Variant::Map::const_iterator i = node.begin(); i !=
node.end(); ++i) {
+ if (i->first == TYPE) {
+ pn_data_put_symbol(data, convert(SUPPORTED_DIST_MODES));
+ pn_data_put_string(data, convert(i->second == TOPIC ? COPY :
MOVE));
+ } else {
+ pn_data_put_symbol(data, convert(i->first));
+ pn_data_put_string(data, convert(i->second.asString()));
+ }
}
- }
- if (node.size()) {
pn_data_exit(data);
}
}
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=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h Sat Mar 23
17:59:50 2013
@@ -40,7 +40,7 @@ class AddressHelper
bool deleteEnabled(CheckMode mode) const;
bool assertEnabled(CheckMode mode) const;
- void setNodeProperties(pn_terminus_t*);
+ void setNodeProperties(pn_terminus_t*, bool dynamic);
const qpid::types::Variant::Map& getNodeProperties() const;
const qpid::types::Variant::Map& getLinkProperties() const;
private:
@@ -49,6 +49,7 @@ class AddressHelper
std::string deletePolicy;
qpid::types::Variant::Map node;
qpid::types::Variant::Map link;
+ std::string name;
bool enabled(const std::string& policy, CheckMode mode) const;
};
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=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Sat Mar
23 17:59:50 2013
@@ -26,6 +26,7 @@
#include "SessionContext.h"
#include "Transport.h"
#include "qpid/messaging/exceptions.h"
+#include "qpid/messaging/AddressImpl.h"
#include "qpid/messaging/Duration.h"
#include "qpid/messaging/Message.h"
#include "qpid/messaging/MessageImpl.h"
@@ -285,34 +286,45 @@ void ConnectionContext::attach(boost::sh
{
lnk->configure();
attach(ssn->session, (pn_link_t*) lnk->sender);
- if (!pn_link_remote_target((pn_link_t*) lnk->sender)) {
+ pn_terminus_t* t = pn_link_remote_target(lnk->sender);
+ if (!pn_terminus_get_address(t)) {
std::string msg("No such target : ");
msg += lnk->getTarget();
+ QPID_LOG(debug, msg);
throw qpid::messaging::NotFound(msg);
+ } else if (AddressImpl::isTemporary(lnk->address)) {
+ lnk->address.setName(pn_terminus_get_address(t));
+ QPID_LOG(debug, "Dynamic target name set to " <<
lnk->address.getName());
}
+ QPID_LOG(debug, "Attach succeeded to " << lnk->getTarget());
}
void ConnectionContext::attach(boost::shared_ptr<SessionContext> ssn,
boost::shared_ptr<ReceiverContext> lnk)
{
lnk->configure();
attach(ssn->session, lnk->receiver, lnk->capacity);
- if (!pn_link_remote_source(lnk->receiver)) {
+ pn_terminus_t* s = pn_link_remote_source(lnk->receiver);
+ if (!pn_terminus_get_address(s)) {
std::string msg("No such source : ");
msg += lnk->getSource();
+ QPID_LOG(debug, msg);
throw qpid::messaging::NotFound(msg);
+ } else if (AddressImpl::isTemporary(lnk->address)) {
+ lnk->address.setName(pn_terminus_get_address(s));
+ QPID_LOG(debug, "Dynamic source name set to " <<
lnk->address.getName());
}
+ QPID_LOG(debug, "Attach succeeded from " << lnk->getSource());
}
void ConnectionContext::attach(pn_session_t* /*session*/, pn_link_t* link, int
credit)
{
qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock);
- QPID_LOG(debug, "Attaching link " << link << ", state=" <<
pn_link_state(link));
pn_link_open(link);
- QPID_LOG(debug, "Link attached " << link << ", state=" <<
pn_link_state(link));
+ QPID_LOG(debug, "Link attach sent for " << link << ", state=" <<
pn_link_state(link));
if (credit) pn_link_flow(link, credit);
wakeupDriver();
while (pn_link_state(link) & PN_REMOTE_UNINIT) {
- QPID_LOG(debug, "waiting for confirmation of link attach for " << link
<< ", state=" << pn_link_state(link));
+ QPID_LOG(debug, "Waiting for confirmation of link attach for " << link
<< ", state=" << pn_link_state(link) << "...");
wait();
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp Sat Mar 23
17:59:50 2013
@@ -20,9 +20,11 @@
*/
#include "qpid/messaging/amqp/ReceiverContext.h"
#include "qpid/messaging/amqp/AddressHelper.h"
+#include "qpid/messaging/AddressImpl.h"
#include "qpid/messaging/Duration.h"
#include "qpid/messaging/Message.h"
#include "qpid/amqp/descriptors.h"
+#include "qpid/log/Statement.h"
extern "C" {
#include <proton/engine.h>
}
@@ -38,7 +40,7 @@ ReceiverContext::ReceiverContext(pn_sess
capacity(0) {}
ReceiverContext::~ReceiverContext()
{
- pn_link_free(receiver);
+ //pn_link_free(receiver);
}
void ReceiverContext::setCapacity(uint32_t c)
@@ -76,7 +78,7 @@ uint32_t ReceiverContext::getUnsettled()
void ReceiverContext::close()
{
-
+ pn_link_close(receiver);
}
const std::string& ReceiverContext::getName() const
@@ -113,11 +115,18 @@ void ReceiverContext::configure() const
}
void ReceiverContext::configure(pn_terminus_t* source) const
{
- pn_terminus_set_address(source, address.getName().c_str());
//dynamic create:
AddressHelper helper(address);
- if (helper.createEnabled(AddressHelper::FOR_RECEIVER)) {
- helper.setNodeProperties(source);
+ if (AddressImpl::isTemporary(address)) {
+ //application expects a name to be generated
+ QPID_LOG(debug, "source is dynamic");
+ helper.setNodeProperties(source, true);
+ } else {
+ pn_terminus_set_address(source, address.getName().c_str());
+ if (helper.createEnabled(AddressHelper::FOR_RECEIVER)) {
+ //application expects name of node to be as specified
+ helper.setNodeProperties(source, false);
+ }
}
// Look specifically for qpid.selector link property and add a filter for
it
@@ -149,6 +158,11 @@ void ReceiverContext::configure(pn_termi
}
}
+Address ReceiverContext::getAddress() const
+{
+ return address;
+}
+
bool ReceiverContext::isClosed() const
{
return false;//TODO
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h Sat Mar 23
17:59:50 2013
@@ -55,10 +55,11 @@ class ReceiverContext
const std::string& getSource() const;
bool isClosed() const;
void configure() const;
+ Address getAddress() const;
private:
friend class ConnectionContext;
const std::string name;
- const Address address;
+ Address address;
pn_link_t* receiver;
uint32_t capacity;
void configure(pn_terminus_t*) const;
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp Sat Mar 23
17:59:50 2013
@@ -103,4 +103,9 @@ bool ReceiverHandle::isClosed() const
return receiver->isClosed();
}
+Address ReceiverHandle::getAddress() const
+{
+ return receiver->getAddress();
+}
+
}}} // namespace qpid::messaging::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h Sat Mar 23
17:59:50 2013
@@ -53,6 +53,7 @@ class ReceiverHandle : public qpid::mess
const std::string& getName() const;
qpid::messaging::Session getSession() const;
bool isClosed() const;
+ Address getAddress() const;
private:
boost::shared_ptr<ConnectionContext> connection;
boost::shared_ptr<SessionContext> session;
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp Sat Mar 23
17:59:50 2013
@@ -21,6 +21,7 @@
#include "qpid/messaging/amqp/SenderContext.h"
#include "qpid/messaging/amqp/EncodedMessage.h"
#include "qpid/messaging/amqp/AddressHelper.h"
+#include "qpid/messaging/AddressImpl.h"
#include "qpid/amqp/descriptors.h"
#include "qpid/amqp/MessageEncoder.h"
#include "qpid/messaging/exceptions.h"
@@ -44,12 +45,12 @@ SenderContext::SenderContext(pn_session_
SenderContext::~SenderContext()
{
- pn_link_free(sender);
+ //pn_link_free(sender);
}
void SenderContext::close()
{
-
+ pn_link_close(sender);
}
void SenderContext::setCapacity(uint32_t c)
@@ -347,11 +348,16 @@ void SenderContext::configure() const
}
void SenderContext::configure(pn_terminus_t* target) const
{
- pn_terminus_set_address(target, address.getName().c_str());
- //dynamic create:
AddressHelper helper(address);
- if (helper.createEnabled(AddressHelper::FOR_SENDER)) {
- helper.setNodeProperties(target);
+ if (AddressImpl::isTemporary(address)) {
+ //application expects a name to be generated
+ helper.setNodeProperties(target, true);
+ } else {
+ pn_terminus_set_address(target, address.getName().c_str());
+ if (helper.createEnabled(AddressHelper::FOR_SENDER)) {
+ //application expects name of node to be as specified
+ helper.setNodeProperties(target, false);
+ }
}
}
@@ -360,4 +366,9 @@ bool SenderContext::settled()
return processUnsettled() == 0;
}
+Address SenderContext::getAddress() const
+{
+ return address;
+}
+
}}} // namespace qpid::messaging::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h Sat Mar 23
17:59:50 2013
@@ -71,12 +71,13 @@ class SenderContext
Delivery* send(const qpid::messaging::Message& message);
void configure() const;
bool settled();
+ Address getAddress() const;
private:
friend class ConnectionContext;
typedef std::deque<Delivery> Deliveries;
const std::string name;
- const qpid::messaging::Address address;
+ qpid::messaging::Address address;
pn_link_t* sender;
int32_t nextId;
Deliveries deliveries;
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp Sat Mar 23
17:59:50 2013
@@ -72,4 +72,9 @@ qpid::messaging::Session SenderHandle::g
return qpid::messaging::Session(new SessionHandle(connection, session));
}
+Address SenderHandle::getAddress() const
+{
+ return sender->getAddress();
+}
+
}}} // namespace qpid::messaging::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h?rev=1460198&r1=1460197&r2=1460198&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h Sat Mar 23
17:59:50 2013
@@ -48,6 +48,7 @@ class SenderHandle : public qpid::messag
uint32_t getUnsettled();
const std::string& getName() const;
Session getSession() const;
+ Address getAddress() const;
private:
boost::shared_ptr<ConnectionContext> connection;
boost::shared_ptr<SessionContext> session;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
