text: /*
   *
   * 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
  