| /* |
| * |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * 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 |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| * |
| */ |
| #include "qpid/broker/amqp/Translation.h" |
| #include "qpid/broker/amqp/Outgoing.h" |
| #include "qpid/broker/amqp_0_10/MessageTransfer.h" |
| #include "qpid/broker/Broker.h" |
| #include "qpid/amqp/Decoder.h" |
| #include "qpid/amqp/descriptors.h" |
| #include "qpid/amqp/MessageEncoder.h" |
| #include "qpid/amqp_0_10/Codecs.h" |
| #include "qpid/types/Variant.h" |
| #include "qpid/types/encodings.h" |
| #include "qpid/framing/MessageTransferBody.h" |
| #include "qpid/log/Statement.h" |
| #include <boost/lexical_cast.hpp> |
| |
| namespace qpid { |
| namespace broker { |
| namespace amqp { |
| namespace { |
| |
| const std::string EMPTY; |
| const std::string FORWARD_SLASH("/"); |
| const std::string TEXT_PLAIN("text/plain"); |
| const std::string SUBJECT_KEY("qpid.subject"); |
| const std::string APP_ID("x-amqp-0-10.app-id"); |
| |
| qpid::framing::ReplyTo translate(const std::string address, Broker* broker) |
| { |
| size_t i = address.find(FORWARD_SLASH); |
| if (i == std::string::npos) { |
| //is it a queue or an exchange? |
| if (broker && broker->getQueues().find(address)) { |
| return qpid::framing::ReplyTo(EMPTY, address); |
| } else if (broker && broker->getExchanges().find(address)) { |
| return qpid::framing::ReplyTo(address, EMPTY); |
| } else { |
| return qpid::framing::ReplyTo(); |
| } |
| } else { |
| return qpid::framing::ReplyTo(i > 0 ? address.substr(0, i) : EMPTY, (i+1) < address.size() ? address.substr(i+1) : EMPTY); |
| } |
| } |
| std::string translate(const qpid::framing::ReplyTo r) |
| { |
| if (r.getExchange().size()) { |
| if (r.getRoutingKey().size()) return r.getExchange() + FORWARD_SLASH + r.getRoutingKey(); |
| else return r.getExchange(); |
| } else return r.getRoutingKey(); |
| } |
| std::string translate(const qpid::amqp::CharSequence& chars) |
| { |
| if (chars.data && chars.size) return std::string(chars.data, chars.size); |
| else return EMPTY; |
| } |
| bool setMessageId(qpid::framing::MessageProperties& m, const qpid::amqp::CharSequence& chars) |
| { |
| if (chars.data && chars.size) { |
| if (chars.size == 16) { |
| m.setMessageId(qpid::framing::Uuid(chars.data)); |
| return true; |
| } else { |
| std::istringstream in(translate(chars)); |
| qpid::framing::Uuid uuid; |
| in >> uuid; |
| if (!in.fail()) { |
| m.setMessageId(uuid); |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| class Properties_0_10 : public qpid::amqp::MessageEncoder::Properties |
| { |
| public: |
| bool hasMessageId() const { return messageProperties && messageProperties->hasMessageId(); } |
| std::string getMessageId() const { return messageProperties ? messageProperties->getMessageId().str() : EMPTY; } |
| bool hasUserId() const { return messageProperties && messageProperties->hasUserId(); } |
| std::string getUserId() const { return messageProperties ? messageProperties->getUserId() : EMPTY; } |
| bool hasTo() const { return getDestination().size() || hasSubject(); } |
| std::string getTo() const { return getDestination().size() ? getDestination() : getSubject(); } |
| bool hasSubject() const |
| { |
| if (getDestination().empty()) { |
| return getApplicationProperties().isSet(SUBJECT_KEY); |
| } else { |
| return deliveryProperties && deliveryProperties->hasRoutingKey(); |
| } |
| } |
| std::string getSubject() const |
| { |
| if (getDestination().empty()) { |
| //message was sent to default exchange, routing key is the queue name |
| return getApplicationProperties().getAsString(SUBJECT_KEY); |
| } else if (deliveryProperties) { |
| return deliveryProperties->getRoutingKey(); |
| } else { |
| return EMPTY; |
| } |
| } |
| bool hasReplyTo() const { return messageProperties && messageProperties->hasReplyTo(); } |
| std::string getReplyTo() const { return messageProperties ? translate(messageProperties->getReplyTo()) : EMPTY; } |
| bool hasCorrelationId() const { return messageProperties && messageProperties->hasCorrelationId(); } |
| std::string getCorrelationId() const { return messageProperties ? messageProperties->getCorrelationId() : EMPTY; } |
| bool hasContentType() const { return messageProperties && messageProperties->hasContentType(); } |
| std::string getContentType() const { return messageProperties ? messageProperties->getContentType() : EMPTY; } |
| bool hasContentEncoding() const { return messageProperties && messageProperties->hasContentEncoding(); } |
| std::string getContentEncoding() const { return messageProperties ? messageProperties->getContentEncoding() : EMPTY; } |
| bool hasAbsoluteExpiryTime() const { return deliveryProperties && deliveryProperties->hasExpiration(); } |
| int64_t getAbsoluteExpiryTime() const { return deliveryProperties ? deliveryProperties->getExpiration() : 0; } |
| bool hasCreationTime() const { return false; } |
| int64_t getCreationTime() const { return 0; } |
| bool hasGroupId() const {return false; } |
| std::string getGroupId() const { return EMPTY; } |
| bool hasGroupSequence() const { return false; } |
| uint32_t getGroupSequence() const { return 0; } |
| bool hasReplyToGroupId() const { return false; } |
| std::string getReplyToGroupId() const { return EMPTY; } |
| |
| const qpid::framing::FieldTable& getApplicationProperties() const { return messageProperties->getApplicationHeaders(); } |
| Properties_0_10(const qpid::broker::amqp_0_10::MessageTransfer& t) : transfer(t), |
| messageProperties(transfer.getProperties<qpid::framing::MessageProperties>()), |
| deliveryProperties(transfer.getProperties<qpid::framing::DeliveryProperties>()) |
| {} |
| private: |
| const qpid::broker::amqp_0_10::MessageTransfer& transfer; |
| const qpid::framing::MessageProperties* messageProperties; |
| const qpid::framing::DeliveryProperties* deliveryProperties; |
| |
| std::string getDestination() const |
| { |
| return transfer.getMethod<qpid::framing::MessageTransferBody>()->getDestination(); |
| } |
| }; |
| } |
| |
| Translation::Translation(const qpid::broker::Message& m, Broker* b) : original(m), broker(b) {} |
| |
| boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> Translation::getTransfer() |
| { |
| boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> t = |
| boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer>(dynamic_cast<const qpid::broker::amqp_0_10::MessageTransfer*>(&original.getEncoding())); |
| if (t) { |
| return t;//no translation required |
| } else { |
| const Message* message = dynamic_cast<const Message*>(&original.getEncoding()); |
| if (message) { |
| //translate 1.0 message into 0-10 |
| boost::intrusive_ptr<qpid::broker::amqp_0_10::MessageTransfer> transfer(new qpid::broker::amqp_0_10::MessageTransfer()); |
| qpid::framing::AMQFrame method((qpid::framing::MessageTransferBody(qpid::framing::ProtocolVersion(), EMPTY, 0, 0))); |
| qpid::framing::AMQFrame header((qpid::framing::AMQHeaderBody())); |
| qpid::framing::AMQFrame content((qpid::framing::AMQContentBody())); |
| method.setEof(false); |
| header.setBof(false); |
| header.setEof(false); |
| content.setBof(false); |
| |
| transfer->getFrames().append(method); |
| transfer->getFrames().append(header); |
| |
| qpid::framing::MessageProperties* props = |
| transfer->getFrames().getHeaders()->get<qpid::framing::MessageProperties>(true); |
| |
| if (message->isTypedBody()) { |
| qpid::types::Variant body = message->getTypedBody(); |
| std::string& data = content.castBody<qpid::framing::AMQContentBody>()->getData(); |
| if (body.getType() == qpid::types::VAR_MAP) { |
| qpid::amqp_0_10::MapCodec::encode(body.asMap(), data); |
| props->setContentType(qpid::amqp_0_10::MapCodec::contentType); |
| } else if (body.getType() == qpid::types::VAR_LIST) { |
| qpid::amqp_0_10::ListCodec::encode(body.asList(), data); |
| props->setContentType(qpid::amqp_0_10::ListCodec::contentType); |
| } else if (body.getType() == qpid::types::VAR_STRING) { |
| data = body.getString(); |
| if (body.getEncoding() == qpid::types::encodings::UTF8 || body.getEncoding() == qpid::types::encodings::ASCII) { |
| props->setContentType(TEXT_PLAIN); |
| } |
| } else { |
| qpid::types::Variant::List container; |
| container.push_back(body); |
| qpid::amqp_0_10::ListCodec::encode(container, data); |
| props->setContentType(qpid::amqp_0_10::ListCodec::contentType); |
| } |
| transfer->getFrames().append(content); |
| props->setContentLength(data.size()); |
| } else { |
| qpid::amqp::CharSequence body = message->getBody(); |
| content.castBody<qpid::framing::AMQContentBody>()->getData().assign(body.data, body.size); |
| transfer->getFrames().append(content); |
| |
| props->setContentLength(body.size); |
| } |
| |
| qpid::amqp::MessageId mid = message->getMessageId(); |
| qpid::framing::Uuid uuid; |
| switch (mid.type) { |
| case qpid::amqp::MessageId::NONE: |
| break; |
| case qpid::amqp::MessageId::UUID: |
| case qpid::amqp::MessageId::BYTES: |
| if (mid.value.bytes.size == 0) break; |
| if (setMessageId(*props, mid.value.bytes)) break; |
| case qpid::amqp::MessageId::ULONG: |
| QPID_LOG(info, "Skipping message id in translation from 1.0 to 0-10 as it is not a UUID"); |
| break; |
| } |
| |
| qpid::amqp::MessageId cid = message->getCorrelationId(); |
| switch (cid.type) { |
| case qpid::amqp::MessageId::NONE: |
| break; |
| case qpid::amqp::MessageId::UUID: |
| assert(cid.value.bytes.size = 16); |
| props->setCorrelationId(qpid::framing::Uuid(cid.value.bytes.data).str()); |
| break; |
| case qpid::amqp::MessageId::BYTES: |
| if (cid.value.bytes.size) { |
| props->setCorrelationId(translate(cid.value.bytes)); |
| } |
| break; |
| case qpid::amqp::MessageId::ULONG: |
| props->setCorrelationId(boost::lexical_cast<std::string>(cid.value.ulong)); |
| break; |
| } |
| if (message->getReplyToAsCharSequence()) props->setReplyTo(translate(message->getReplyTo(), broker)); |
| if (message->getContentType()) props->setContentType(translate(message->getContentType())); |
| if (message->getContentEncoding()) props->setContentEncoding(translate(message->getContentEncoding())); |
| props->setUserId(message->getUserId()); |
| // TODO: FieldTable applicationHeaders; |
| qpid::amqp::CharSequence ap = message->getApplicationProperties(); |
| if (ap) { |
| qpid::amqp::Decoder d(ap.data, ap.size); |
| qpid::amqp_0_10::translate(d.readMap(), props->getApplicationHeaders()); |
| std::string appid = props->getApplicationHeaders().getAsString(APP_ID); |
| if (!appid.empty()) { |
| props->setAppId(appid); |
| } |
| } |
| |
| qpid::framing::DeliveryProperties* dp = |
| transfer->getFrames().getHeaders()->get<qpid::framing::DeliveryProperties>(true); |
| dp->setPriority(message->getPriority()); |
| if (message->isPersistent()) dp->setDeliveryMode(2); |
| if (message->getRoutingKey().size()) { |
| if (message->getRoutingKey().size() > std::numeric_limits<uint8_t>::max()) { |
| //have to truncate routing key as it is specified to be a str8 |
| dp->setRoutingKey(message->getRoutingKey().substr(0,std::numeric_limits<uint8_t>::max())); |
| } else { |
| dp->setRoutingKey(message->getRoutingKey()); |
| } |
| props->getApplicationHeaders().setString(SUBJECT_KEY, message->getRoutingKey()); |
| } |
| |
| return transfer.get(); |
| } else { |
| throw qpid::Exception("Could not write message data in AMQP 0-10 format"); |
| } |
| } |
| } |
| |
| void Translation::write(OutgoingFromQueue& out) |
| { |
| const Message* message = dynamic_cast<const Message*>(original.getPersistentContext().get()); |
| //persistent context will contain any newly added annotations |
| if (!message) message = dynamic_cast<const Message*>(&original.getEncoding()); |
| if (message) { |
| //write annotations |
| qpid::amqp::CharSequence deliveryAnnotations = message->getDeliveryAnnotations(); |
| qpid::amqp::CharSequence messageAnnotations = message->getMessageAnnotations(); |
| if (deliveryAnnotations.size) out.write(deliveryAnnotations.data, deliveryAnnotations.size); |
| if (messageAnnotations.size) out.write(messageAnnotations.data, messageAnnotations.size); |
| //write bare message |
| qpid::amqp::CharSequence bareMessage = message->getBareMessage(); |
| if (bareMessage.size) out.write(bareMessage.data, bareMessage.size); |
| //write footer: |
| qpid::amqp::CharSequence footer = message->getFooter(); |
| if (footer.size) out.write(footer.data, footer.size); |
| } else { |
| const qpid::broker::amqp_0_10::MessageTransfer* transfer = dynamic_cast<const qpid::broker::amqp_0_10::MessageTransfer*>(&original.getEncoding()); |
| if (transfer) { |
| Properties_0_10 properties(*transfer); |
| qpid::types::Variant::Map applicationProperties; |
| qpid::amqp_0_10::translate(properties.getApplicationProperties(), applicationProperties); |
| if (properties.getContentType() == qpid::amqp_0_10::MapCodec::contentType) { |
| qpid::types::Variant::Map content; |
| qpid::amqp_0_10::MapCodec::decode(transfer->getContent(), content); |
| size_t size = qpid::amqp::MessageEncoder::getEncodedSize(properties); |
| size += qpid::amqp::MessageEncoder::getEncodedSize(applicationProperties, true) + 3;/*descriptor*/ |
| size += qpid::amqp::MessageEncoder::getEncodedSize(content, true) + 3/*descriptor*/; |
| std::vector<char> buffer(size); |
| qpid::amqp::MessageEncoder encoder(&buffer[0], buffer.size()); |
| encoder.writeProperties(properties); |
| encoder.writeApplicationProperties(applicationProperties); |
| encoder.writeMap(content, &qpid::amqp::message::AMQP_VALUE); |
| out.write(&buffer[0], encoder.getPosition()); |
| } else if (properties.getContentType() == qpid::amqp_0_10::ListCodec::contentType) { |
| qpid::types::Variant::List content; |
| qpid::amqp_0_10::ListCodec::decode(transfer->getContent(), content); |
| size_t size = qpid::amqp::MessageEncoder::getEncodedSize(properties); |
| size += qpid::amqp::MessageEncoder::getEncodedSize(applicationProperties, true) + 3;/*descriptor*/ |
| size += qpid::amqp::MessageEncoder::getEncodedSize(content, true) + 3/*descriptor*/; |
| std::vector<char> buffer(size); |
| qpid::amqp::MessageEncoder encoder(&buffer[0], buffer.size()); |
| encoder.writeProperties(properties); |
| encoder.writeApplicationProperties(applicationProperties); |
| encoder.writeList(content, &qpid::amqp::message::AMQP_VALUE); |
| out.write(&buffer[0], encoder.getPosition()); |
| } else { |
| std::string content = transfer->getContent(); |
| size_t size = qpid::amqp::MessageEncoder::getEncodedSize(properties, applicationProperties, content); |
| std::vector<char> buffer(size); |
| qpid::amqp::MessageEncoder encoder(&buffer[0], buffer.size()); |
| encoder.writeProperties(properties); |
| encoder.writeApplicationProperties(applicationProperties); |
| if (content.size()) encoder.writeBinary(content, &qpid::amqp::message::DATA); |
| out.write(&buffer[0], encoder.getPosition()); |
| } |
| } else { |
| QPID_LOG(error, "Could not write message data in AMQP 1.0 format"); |
| } |
| } |
| } |
| |
| }}} // namespace qpid::broker::amqp |
