You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2009/11/23 12:20:19 UTC
svn commit: r883304 - in /qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10:
AddressResolution.cpp OutgoingMessage.cpp OutgoingMessage.h SenderImpl.cpp
Author: gsim
Date: Mon Nov 23 11:20:18 2009
New Revision: 883304
URL: http://svn.apache.org/viewvc?rev=883304&view=rev
Log:
QPID-664: Set subject in application headers when specified
Modified:
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp?rev=883304&r1=883303&r2=883304&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp Mon Nov 23 11:20:18 2009
@@ -224,7 +224,6 @@
void send(qpid::client::AsyncSession& session, const std::string& name, OutgoingMessage& message);
void cancel(qpid::client::AsyncSession& session, const std::string& name);
private:
- const std::string defaultSubject;
};
class QueueSink : public Queue, public MessageSink
@@ -406,9 +405,8 @@
const Variant& filter = address.getOption(FILTER);
if (!filter.isVoid()) {
- //TODO: if both subject _and_ filter are specified,
- //combine in some way; for now we just ignore the
- //subject in that case.
+ //TODO: if both subject _and_ filter are specified, combine in
+ //some way; for now we just ignore the subject in that case.
bind(filter);
} else if (address.hasSubject()) {
//Note: This will not work for headers- or xml- exchange;
@@ -459,9 +457,7 @@
Subscription::Binding::Binding(const std::string& e, const std::string& k, const FieldTable& o):
exchange(e), key(k), options(o) {}
-void convert(qpid::messaging::Message& from, qpid::client::Message& to);
-
-ExchangeSink::ExchangeSink(const Address& address) : Exchange(address), defaultSubject(address.getSubject()) {}
+ExchangeSink::ExchangeSink(const Address& address) : Exchange(address) {}
void ExchangeSink::declare(qpid::client::AsyncSession& session, const std::string&)
{
@@ -471,9 +467,7 @@
void ExchangeSink::send(qpid::client::AsyncSession& session, const std::string&, OutgoingMessage& m)
{
- if (m.message.getDeliveryProperties().getRoutingKey().empty() && !defaultSubject.empty()) {
- m.message.getDeliveryProperties().setRoutingKey(defaultSubject);
- }
+ m.message.getDeliveryProperties().setRoutingKey(m.getSubject());
m.status = session.messageTransfer(arg::destination=name, arg::content=m.message);
}
@@ -500,21 +494,6 @@
checkDelete(session, FOR_SENDER);
}
-void convert(qpid::messaging::Message& from, qpid::client::Message& to)
-{
- //TODO: need to avoid copying as much as possible
- to.setData(from.getContent());
- to.getDeliveryProperties().setRoutingKey(from.getSubject());
- //TODO: set other delivery properties
- to.getMessageProperties().setContentType(from.getContentType());
- const Address& address = from.getReplyTo();
- if (!address.getName().empty()) {
- to.getMessageProperties().setReplyTo(AddressResolution::convert(address));
- }
- translate(from.getHeaders(), to.getMessageProperties().getApplicationHeaders());
- //TODO: set other message properties
-}
-
Address AddressResolution::convert(const qpid::framing::ReplyTo& rt)
{
Address address;
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp?rev=883304&r1=883303&r2=883304&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.cpp Mon Nov 23 11:20:18 2009
@@ -48,4 +48,20 @@
//TODO: set other delivery properties
}
+namespace {
+const std::string SUBJECT("subject");
+}
+
+void OutgoingMessage::setSubject(const std::string& subject)
+{
+ if (!subject.empty()) {
+ message.getMessageProperties().getApplicationHeaders().setString(SUBJECT, subject);
+ }
+}
+
+std::string OutgoingMessage::getSubject() const
+{
+ return message.getMessageProperties().getApplicationHeaders().getAsString(SUBJECT);
+}
+
}}} // namespace qpid::client::amqp0_10
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h?rev=883304&r1=883303&r2=883304&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/OutgoingMessage.h Mon Nov 23 11:20:18 2009
@@ -37,6 +37,8 @@
qpid::client::Completion status;
void convert(const qpid::messaging::Message&);
+ void setSubject(const std::string& subject);
+ std::string getSubject() const;
};
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=883304&r1=883303&r2=883304&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 Mon Nov 23 11:20:18 2009
@@ -98,6 +98,7 @@
//TODO: make recording for replay optional (would still want to track completion however)
std::auto_ptr<OutgoingMessage> msg(new OutgoingMessage());
msg->convert(m);
+ msg->setSubject(m.getSubject().empty() ? address.getSubject() : m.getSubject());
outgoing.push_back(msg.release());
sink->send(session, name, outgoing.back());
}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org