/*
 *
 * 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/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/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("/");

std::string translate(const qpid::framing::ReplyTo r)
{
    if (r.hasExchange()) {
        if (r.hasRoutingKey()) 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 { return deliveryProperties && getDestination().size() && deliveryProperties->hasRoutingKey(); }
    std::string getSubject() const { return deliveryProperties && getDestination().size() ? deliveryProperties->getRoutingKey() : 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() { 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) : original(m) {}


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::amqp::CharSequence body = message->getBody();
            content.castBody<qpid::framing::AMQContentBody>()->getData().assign(body.data, body.size);
            transfer->getFrames().append(content);

            qpid::framing::MessageProperties* props =
                transfer->getFrames().getHeaders()->get<qpid::framing::MessageProperties>(true);
            props->setContentLength(body.size);

            qpid::amqp::MessageId mid = message->getMessageId();
            qpid::framing::Uuid uuid;
            switch (mid.type) {
              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::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;
            }
            // TODO: ReplyTo - there is no way to reliably determine
            // the type of the node from just its name, unless we
            // query the brokers registries

            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());
            }

            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()) dp->setRoutingKey(message->getRoutingKey());

            return transfer.get();
        } else {
            throw qpid::Exception("Could not write message data in AMQP 0-10 format");
        }
    }
}

void Translation::write(Outgoing& out)
{
    const Message* message = dynamic_cast<const Message*>(&original.getEncoding());
    if (message) {
        //write annotations
        //TODO: merge in any newly added 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);
            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);
            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