summaryrefslogtreecommitdiff
path: root/cpp/src/qpid/messaging
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid/messaging')
-rw-r--r--cpp/src/qpid/messaging/amqp/ConnectionContext.cpp7
-rw-r--r--cpp/src/qpid/messaging/amqp/ReceiverContext.cpp37
-rw-r--r--cpp/src/qpid/messaging/amqp/ReceiverContext.h9
-rw-r--r--cpp/src/qpid/messaging/amqp/SenderContext.cpp29
-rw-r--r--cpp/src/qpid/messaging/amqp/SenderContext.h7
-rw-r--r--cpp/src/qpid/messaging/amqp/SessionContext.cpp4
6 files changed, 67 insertions, 26 deletions
diff --git a/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp b/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
index 499b1ae35d..2f1f39cb68 100644
--- a/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
+++ b/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
@@ -262,10 +262,9 @@ void ConnectionContext::attach(boost::shared_ptr<SessionContext> ssn, boost::sha
void ConnectionContext::attach(boost::shared_ptr<SessionContext> ssn, boost::shared_ptr<ReceiverContext> lnk)
{
- pn_terminus_t* source = pn_link_source((pn_link_t*) lnk->receiver);
- pn_terminus_set_address(source, lnk->getSource().c_str());
- attach(ssn->session, (pn_link_t*) lnk->receiver, lnk->capacity);
- if (!pn_link_remote_source((pn_link_t*) lnk->receiver)) {
+ lnk->configure();
+ attach(ssn->session, lnk->receiver, lnk->capacity);
+ if (!pn_link_remote_source(lnk->receiver)) {
std::string msg("No such source : ");
msg += lnk->getSource();
throw qpid::messaging::NotFound(msg);
diff --git a/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp b/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
index 7e41e727cc..33ddabfc75 100644
--- a/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
+++ b/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
@@ -29,10 +29,10 @@ namespace qpid {
namespace messaging {
namespace amqp {
//TODO: proper conversion to wide string for address
-ReceiverContext::ReceiverContext(pn_session_t* session, const std::string& n, const std::string& s)
+ReceiverContext::ReceiverContext(pn_session_t* session, const std::string& n, const qpid::messaging::Address& a)
: name(n),
- source(s),
- receiver(pn_receiver(session, source.c_str())),
+ address(a),
+ receiver(pn_receiver(session, name.c_str())),
capacity(0) {}
ReceiverContext::~ReceiverContext()
{
@@ -84,7 +84,36 @@ const std::string& ReceiverContext::getName() const
const std::string& ReceiverContext::getSource() const
{
- return source;
+ return address.getName();
+}
+namespace {
+pn_bytes_t convert(const std::string& s)
+{
+ pn_bytes_t result;
+ result.start = const_cast<char*>(s.data());
+ result.size = s.size();
+ return result;
+}
+}
+
+void ReceiverContext::configure() const
+{
+ configure(pn_link_source(receiver));
+}
+void ReceiverContext::configure(pn_terminus_t* source) const
+{
+ pn_terminus_set_address(source, address.getName().c_str());
+
+ pn_data_t* filter = pn_terminus_filter(source);
+ pn_data_put_map(filter);
+ pn_data_enter(filter);
+ pn_data_put_symbol(filter, convert("subject"));
+ pn_data_put_described(filter);
+ pn_data_enter(filter);
+ pn_data_put_ulong(filter, 0x0000468C00000001/*LEGACY_TOPIC_FILTER*/);
+ pn_data_put_string(filter, convert(address.getSubject()));
+ pn_data_exit(filter);
+ pn_data_exit(filter);
}
bool ReceiverContext::isClosed() const
diff --git a/cpp/src/qpid/messaging/amqp/ReceiverContext.h b/cpp/src/qpid/messaging/amqp/ReceiverContext.h
index 0a6f363228..34ecdda6be 100644
--- a/cpp/src/qpid/messaging/amqp/ReceiverContext.h
+++ b/cpp/src/qpid/messaging/amqp/ReceiverContext.h
@@ -21,11 +21,13 @@
* under the License.
*
*/
+#include "qpid/messaging/Address.h"
#include <string>
#include "qpid/sys/IntegerTypes.h"
struct pn_link_t;
struct pn_session_t;
+struct pn_terminus_t;
namespace qpid {
namespace messaging {
@@ -41,22 +43,25 @@ namespace amqp {
class ReceiverContext
{
public:
- ReceiverContext(pn_session_t* session, const std::string& name, const std::string& source);
+ ReceiverContext(pn_session_t* session, const std::string& name, const qpid::messaging::Address& source);
~ReceiverContext();
void setCapacity(uint32_t);
uint32_t getCapacity();
uint32_t getAvailable();
uint32_t getUnsettled();
+ void attach();
void close();
const std::string& getName() const;
const std::string& getSource() const;
bool isClosed() const;
+ void configure() const;
private:
friend class ConnectionContext;
const std::string name;
- const std::string source;
+ const Address address;
pn_link_t* receiver;
uint32_t capacity;
+ void configure(pn_terminus_t*) const;
};
}}} // namespace qpid::messaging::amqp
diff --git a/cpp/src/qpid/messaging/amqp/SenderContext.cpp b/cpp/src/qpid/messaging/amqp/SenderContext.cpp
index 1306a314be..02f7bdec1c 100644
--- a/cpp/src/qpid/messaging/amqp/SenderContext.cpp
+++ b/cpp/src/qpid/messaging/amqp/SenderContext.cpp
@@ -36,10 +36,10 @@ namespace qpid {
namespace messaging {
namespace amqp {
//TODO: proper conversion to wide string for address
-SenderContext::SenderContext(pn_session_t* session, const std::string& n, const std::string& t)
+SenderContext::SenderContext(pn_session_t* session, const std::string& n, const qpid::messaging::Address& a)
: name(n),
- target(t),
- sender(pn_sender(session, target.c_str())), capacity(1000) {}
+ address(a),
+ sender(pn_sender(session, n.c_str())), capacity(1000) {}
SenderContext::~SenderContext()
{
@@ -74,7 +74,7 @@ const std::string& SenderContext::getName() const
const std::string& SenderContext::getTarget() const
{
- return target;
+ return address.getName();
}
SenderContext::Delivery* SenderContext::send(const qpid::messaging::Message& message)
@@ -82,7 +82,7 @@ SenderContext::Delivery* SenderContext::send(const qpid::messaging::Message& mes
if (processUnsettled() < capacity) {
deliveries.push_back(Delivery(nextId++));
Delivery& delivery = deliveries.back();
- delivery.encode(MessageImplAccess::get(message));
+ delivery.encode(MessageImplAccess::get(message), address);
delivery.send(sender);
return &delivery;
} else {
@@ -135,7 +135,7 @@ const std::string EMPTY;
class PropertiesAdapter : public qpid::amqp::MessageEncoder::Properties
{
public:
- PropertiesAdapter(const qpid::messaging::MessageImpl& impl) : msg(impl) {}
+ PropertiesAdapter(const qpid::messaging::MessageImpl& impl, const std::string& s) : msg(impl), subject(s) {}
bool hasMessageId() const
{
return getMessageId().size();
@@ -167,12 +167,12 @@ class PropertiesAdapter : public qpid::amqp::MessageEncoder::Properties
bool hasSubject() const
{
- return getSubject().size();
+ return subject.size() || getSubject().size();
}
std::string getSubject() const
{
- return msg.getSubject();
+ return subject.size() ? subject : msg.getSubject();
}
bool hasReplyTo() const
@@ -266,16 +266,23 @@ class PropertiesAdapter : public qpid::amqp::MessageEncoder::Properties
}
private:
const qpid::messaging::MessageImpl& msg;
+ const std::string subject;
};
+
+bool changedSubject(const qpid::messaging::MessageImpl& msg, const qpid::messaging::Address& address)
+{
+ return address.getSubject().size() && address.getSubject() != msg.getSubject();
+}
+
}
SenderContext::Delivery::Delivery(int32_t i) : id(i), token(0) {}
-void SenderContext::Delivery::encode(const qpid::messaging::MessageImpl& msg)
+void SenderContext::Delivery::encode(const qpid::messaging::MessageImpl& msg, const qpid::messaging::Address& address)
{
boost::shared_ptr<const EncodedMessage> original = msg.getEncoded();
- if (original) { //still have the content as received, send at least the bare message unaltered
+ if (original && !changedSubject(msg, address)) { //still have the content as received, send at least the bare message unaltered
//do we need to alter the header? are durable, priority, ttl, first-acquirer, delivery-count different from what was received?
if (original->hasHeaderChanged(msg)) {
//since as yet have no annotations, just write the revised header then the rest of the message as received
@@ -293,7 +300,7 @@ void SenderContext::Delivery::encode(const qpid::messaging::MessageImpl& msg)
}
} else {
HeaderAdapter header(msg);
- PropertiesAdapter properties(msg);
+ PropertiesAdapter properties(msg, address.getSubject());
//compute size:
encoded.resize(qpid::amqp::MessageEncoder::getEncodedSize(header, properties, msg.getHeaders(), msg.getBytes()));
QPID_LOG(debug, "Sending message, buffer is " << encoded.getSize() << " bytes")
diff --git a/cpp/src/qpid/messaging/amqp/SenderContext.h b/cpp/src/qpid/messaging/amqp/SenderContext.h
index 82c5e6dab9..bc73d199e7 100644
--- a/cpp/src/qpid/messaging/amqp/SenderContext.h
+++ b/cpp/src/qpid/messaging/amqp/SenderContext.h
@@ -25,6 +25,7 @@
#include <string>
#include <vector>
#include "qpid/sys/IntegerTypes.h"
+#include "qpid/messaging/Address.h"
#include "qpid/messaging/amqp/EncodedMessage.h"
struct pn_delivery_t;
@@ -48,7 +49,7 @@ class SenderContext
{
public:
Delivery(int32_t id);
- void encode(const qpid::messaging::MessageImpl& message);
+ void encode(const qpid::messaging::MessageImpl& message, const qpid::messaging::Address&);
void send(pn_link_t*);
bool accepted();
private:
@@ -57,7 +58,7 @@ class SenderContext
EncodedMessage encoded;
};
- SenderContext(pn_session_t* session, const std::string& name, const std::string& target);
+ SenderContext(pn_session_t* session, const std::string& name, const qpid::messaging::Address& target);
~SenderContext();
void close();
void setCapacity(uint32_t);
@@ -71,7 +72,7 @@ class SenderContext
typedef std::deque<Delivery> Deliveries;
const std::string name;
- const std::string target;
+ const qpid::messaging::Address address;
pn_link_t* sender;
int32_t nextId;
Deliveries deliveries;
diff --git a/cpp/src/qpid/messaging/amqp/SessionContext.cpp b/cpp/src/qpid/messaging/amqp/SessionContext.cpp
index 9908b16443..8b3feb129a 100644
--- a/cpp/src/qpid/messaging/amqp/SessionContext.cpp
+++ b/cpp/src/qpid/messaging/amqp/SessionContext.cpp
@@ -49,7 +49,7 @@ boost::shared_ptr<SenderContext> SessionContext::createSender(const qpid::messag
for (SenderMap::const_iterator i = senders.find(name); i != senders.end(); i = senders.find(name)) {
name = (boost::format("%1%_%2%") % address.getName() % ++count).str();
}
- boost::shared_ptr<SenderContext> s(new SenderContext(session, name, address.str()));
+ boost::shared_ptr<SenderContext> s(new SenderContext(session, name, address));
senders[name] = s;
return s;
}
@@ -62,7 +62,7 @@ boost::shared_ptr<ReceiverContext> SessionContext::createReceiver(const qpid::me
for (ReceiverMap::const_iterator i = receivers.find(name); i != receivers.end(); i = receivers.find(name)) {
name = (boost::format("%1%_%2%") % address.getName() % ++count).str();
}
- boost::shared_ptr<ReceiverContext> r(new ReceiverContext(session, name, address.str()));
+ boost::shared_ptr<ReceiverContext> r(new ReceiverContext(session, name, address));
receivers[name] = r;
return r;
}
View Lorry Controller Status