/*
 *
 * 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 "Session.h"
#include "Incoming.h"
#include "Outgoing.h"
#include "Message.h"
#include "Connection.h"
#include "Domain.h"
#include "Exception.h"
#include "Interconnects.h"
#include "Relay.h"
#include "Topic.h"
#include "qpid/amqp/descriptors.h"
#include "qpid/broker/Broker.h"
#include "qpid/broker/DeliverableMessage.h"
#include "qpid/broker/Exchange.h"
#include "qpid/broker/DirectExchange.h"
#include "qpid/broker/TopicExchange.h"
#include "qpid/broker/FanOutExchange.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/QueueCursor.h"
#include "qpid/broker/Selector.h"
#include "qpid/broker/TopicExchange.h"
#include "qpid/broker/amqp/Filter.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/framing/FieldTable.h"
#include "qpid/framing/MessageTransferBody.h"
#include "qpid/log/Statement.h"
#include "qpid/amqp_0_10/Codecs.h"
#include <boost/intrusive_ptr.hpp>
#include <boost/format.hpp>
#include <map>
#include <sstream>
extern "C" {
#include <proton/engine.h>
}

namespace qpid {
namespace broker {
namespace amqp {

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;
}
std::string convert(pn_bytes_t in)
{
    return std::string(in.start, in.size);
}
//capabilities
const std::string CREATE_ON_DEMAND("create-on-demand");
const std::string DURABLE("durable");
const std::string QUEUE("queue");
const std::string TOPIC("topic");
const std::string DIRECT_FILTER("legacy-amqp-direct-binding");
const std::string TOPIC_FILTER("legacy-amqp-topic-binding");
const std::string SHARED("shared");

void writeCapabilities(pn_data_t* out, const std::vector<std::string>& supported)
{
    if (supported.size() == 1) {
        pn_data_put_symbol(out, convert(supported.front()));
    } else if (supported.size() > 1) {
        pn_data_put_array(out, false, PN_SYMBOL);
        pn_data_enter(out);
        for (std::vector<std::string>::const_iterator i = supported.begin(); i != supported.end(); ++i) {
            pn_data_put_symbol(out, convert(*i));
        }
        pn_data_exit(out);
    }
}

template <class F>
void readCapabilities(pn_data_t* data, F f)
{
    pn_data_rewind(data);
    if (pn_data_next(data)) {
        pn_type_t type = pn_data_type(data);
        if (type == PN_ARRAY) {
            pn_data_enter(data);
            while (pn_data_next(data)) {
	      std::string s = convert(pn_data_get_symbol(data));
	      f(s);
            }
            pn_data_exit(data);
        } else if (type == PN_SYMBOL) {
	  std::string s = convert(pn_data_get_symbol(data));
	  f(s);
        } else {
            QPID_LOG(error, "Skipping capabilities field of type " << pn_type_name(type));
        }
    }
}

void matchCapability(const std::string& name, bool* result, const std::string& s)
{
    if (s == name) *result = true;
}

bool is_capability_requested(const std::string& name, pn_data_t* capabilities)
{
    bool result(false);
    readCapabilities(capabilities, boost::bind(&matchCapability, name, &result, _1));
    return result;
}

void collectQueueCapabilities(boost::shared_ptr<Queue> node, std::vector<std::string>* supported, const std::string& s)
{
    if (s == DURABLE) {
        if (node->isDurable()) supported->push_back(s);
    } else if (s == CREATE_ON_DEMAND || s == QUEUE || s == DIRECT_FILTER || s == TOPIC_FILTER) {
        supported->push_back(s);
    }
}

void collectExchangeCapabilities(boost::shared_ptr<Exchange> node, std::vector<std::string>* supported, const std::string& s)
{
    if (s == DURABLE) {
        if (node->isDurable()) supported->push_back(s);
    } else if (s == SHARED) {
        supported->push_back(s);
    } else if (s == CREATE_ON_DEMAND || s == TOPIC) {
        supported->push_back(s);
    } else if (s == DIRECT_FILTER) {
        if (node->getType() == DirectExchange::typeName) supported->push_back(s);
    } else if (s == TOPIC_FILTER) {
        if (node->getType() == TopicExchange::typeName) supported->push_back(s);
    }
}

void setCapabilities(pn_data_t* in, pn_data_t* out, boost::shared_ptr<Queue> node)
{
    std::vector<std::string> supported;
    readCapabilities(in, boost::bind(&collectQueueCapabilities, node, &supported, _1));
    writeCapabilities(out, supported);
}

void setCapabilities(pn_data_t* in, pn_data_t* out, boost::shared_ptr<Exchange> node)
{
    std::vector<std::string> supported;
    readCapabilities(in, boost::bind(&collectExchangeCapabilities, node, &supported, _1));
    writeCapabilities(out, supported);
}

}

class IncomingToQueue : public DecodingIncoming
{
  public:
    IncomingToQueue(Broker& b, Session& p, boost::shared_ptr<qpid::broker::Queue> q, pn_link_t* l, const std::string& source, bool icl)
        : DecodingIncoming(l, b, p, source, q->getName(), pn_link_name(l)), queue(q), isControllingLink(icl)
    {
        queue->markInUse(isControllingLink);
    }
    ~IncomingToQueue() { queue->releaseFromUse(isControllingLink); }
    void handle(qpid::broker::Message& m);
  private:
    boost::shared_ptr<qpid::broker::Queue> queue;
    bool isControllingLink;
};

class IncomingToExchange : public DecodingIncoming
{
  public:
    IncomingToExchange(Broker& b, Session& p, boost::shared_ptr<qpid::broker::Exchange> e, pn_link_t* l, const std::string& source)
        : DecodingIncoming(l, b, p, source, e->getName(), pn_link_name(l)), exchange(e), authorise(p.getAuthorise()) {}
    void handle(qpid::broker::Message& m);
  private:
    boost::shared_ptr<qpid::broker::Exchange> exchange;
    Authorise& authorise;
};

Session::Session(pn_session_t* s, Connection& c, qpid::sys::OutputControl& o)
    : ManagedSession(c.getBroker(), c, (boost::format("%1%") % s).str()), session(s), connection(c), out(o), deleted(false),
      authorise(connection.getUserId(), connection.getBroker().getAcl()) {}


Session::ResolvedNode Session::resolve(const std::string name, pn_terminus_t* terminus, bool incoming)
{
    ResolvedNode node;
    node.exchange = connection.getBroker().getExchanges().find(name);
    node.queue = connection.getBroker().getQueues().find(name);
    node.topic = connection.getTopics().get(name);
    bool createOnDemand = is_capability_requested(CREATE_ON_DEMAND, pn_terminus_capabilities(terminus));
    //Strictly speaking, properties should only be specified when the
    //terminus is dynamic. However we will not enforce that here. If
    //properties are set on the attach request, we will set them on
    //our reply. This allows the 'create' and 'assert' options in the
    //qpid messaging API to be implemented over 1.0.
    node.properties.read(pn_terminus_properties(terminus));

    if (node.topic) node.exchange = node.topic->getExchange();
    if (node.exchange && !node.queue && createOnDemand) {
        if (!node.properties.getExchangeType().empty() && node.properties.getExchangeType() != node.exchange->getType()) {
            //emulate 0-10 exchange-declare behaviour
            throw Exception(qpid::amqp::error_conditions::PRECONDITION_FAILED, "Exchange of different type already exists");
        }
    }
    if (!node.queue && !node.exchange) {
        if (pn_terminus_is_dynamic(terminus)  || createOnDemand) {
            //is it a queue or an exchange?
            if (node.properties.isQueue()) {
                node.queue = connection.getBroker().createQueue(name, node.properties.getQueueSettings(), this, node.properties.getAlternateExchange(), connection.getUserId(), connection.getId()).first;
            } else {
                qpid::framing::FieldTable args;
                qpid::amqp_0_10::translate(node.properties.getProperties(), args);
                node.exchange = connection.getBroker().createExchange(name, node.properties.getExchangeType(), node.properties.isDurable(), node.properties.getAlternateExchange(),
                                                      args, connection.getUserId(), connection.getId()).first;
            }
        } else {
            size_t i = name.find('@');
            if (i != std::string::npos && (i+1) < name.length()) {
                std::string domain = name.substr(i+1);
                std::string local = name.substr(0, i);
                std::string id = (boost::format("%1%-%2%") % name % qpid::types::Uuid(true).str()).str();
                //does this domain exist?
                boost::shared_ptr<Domain> d = connection.getInterconnects().findDomain(domain);
                if (d) {
                    node.relay = boost::shared_ptr<Relay>(new Relay(1000));
                    if (incoming) {
                        d->connect(false, id, name, local, connection, node.relay);
                    } else {
                        d->connect(true, id, local, name, connection, node.relay);
                    }
                }
            }
        }
    } else if (node.queue && node.topic) {
        QPID_LOG_CAT(warning, protocol, "Ambiguous node name; " << name << " could be queue or topic, assuming topic");
        node.queue.reset();
    } else if (node.queue && node.exchange) {
        QPID_LOG_CAT(warning, protocol, "Ambiguous node name; " << name << " could be queue or exchange, assuming queue");
        node.exchange.reset();
    }

    if (node.properties.isExclusive() && node.queue && node.queue->setExclusiveOwner(this)) {
        exclusiveQueues.insert(node.queue);
    }

    return node;
}

std::string Session::generateName(pn_link_t* link)
{
    std::stringstream s;
    if (connection.getContainerId().empty()) {
        s << qpid::types::Uuid(true);
    } else {
        s << connection.getContainerId();
    }
    s << "_" << pn_link_name(link);
    return s.str();
}

std::string Session::qualifyName(const std::string& name)
{
    if (connection.getDomain().empty()) {
        return name;
    } else {
        std::stringstream s;
        s << name << "@" << connection.getDomain();
        return s.str();
    }
}

void Session::attach(pn_link_t* link)
{
    if (pn_link_is_sender(link)) {
        pn_terminus_t* source = pn_link_remote_source(link);
        //i.e a subscription
        std::string name;
        if (pn_terminus_get_type(source) == PN_UNSPECIFIED) {
            throw Exception(qpid::amqp::error_conditions::PRECONDITION_FAILED, "No source specified!");
        } else if (pn_terminus_is_dynamic(source)) {
            name = generateName(link);
            QPID_LOG(debug, "Received attach request for outgoing link from " << name);
            pn_terminus_set_address(pn_link_source(link), qualifyName(name).c_str());
        } else {
            name = pn_terminus_get_address(source);
            QPID_LOG(debug, "Received attach request for outgoing link from " << name);
            pn_terminus_set_address(pn_link_source(link), name.c_str());
        }

        setupOutgoing(link, source, name);
    } else {
        pn_terminus_t* target = pn_link_remote_target(link);
        std::string name;
        if (pn_terminus_get_type(target) == PN_UNSPECIFIED) {
            throw Exception(qpid::amqp::error_conditions::PRECONDITION_FAILED, "No target specified!");
        } else if (pn_terminus_is_dynamic(target)) {
            name = generateName(link);
            QPID_LOG(debug, "Received attach request for incoming link to " << name);
            pn_terminus_set_address(pn_link_target(link), qualifyName(name).c_str());
        } else {
            name  = pn_terminus_get_address(target);
            QPID_LOG(debug, "Received attach request for incoming link to " << name);
            pn_terminus_set_address(pn_link_target(link), name.c_str());
        }

        setupIncoming(link, target, name);
    }
}

void Session::setupIncoming(pn_link_t* link, pn_terminus_t* target, const std::string& name)
{
    ResolvedNode node = resolve(name, target, true);
    //set capabilities
    if (node.queue) {
        setCapabilities(pn_terminus_capabilities(target), pn_terminus_capabilities(pn_link_target(link)), node.queue);
        authorise.incoming(node.queue);
        node.properties.write(pn_terminus_properties(pn_link_target(link)), node.queue);
    } else if (node.exchange) {
        setCapabilities(pn_terminus_capabilities(target), pn_terminus_capabilities(pn_link_target(link)), node.exchange);
        authorise.incoming(node.exchange);
        node.properties.write(pn_terminus_properties(pn_link_target(link)), node.exchange);
    }

    const char* sourceAddress = pn_terminus_get_address(pn_link_remote_source(link));
    if (!sourceAddress) {
        sourceAddress = pn_terminus_get_address(pn_link_source(link));
    }
    std::string source;
    if (sourceAddress) {
        source = sourceAddress;
    }
    if (node.queue) {
        boost::shared_ptr<Incoming> q(new IncomingToQueue(connection.getBroker(), *this, node.queue, link, source, node.properties.trackControllingLink()));
        incoming[link] = q;
    } else if (node.exchange) {
        boost::shared_ptr<Incoming> e(new IncomingToExchange(connection.getBroker(), *this, node.exchange, link, source));
        incoming[link] = e;
    } else if (node.relay) {
        boost::shared_ptr<Incoming> in(new IncomingToRelay(link, connection.getBroker(), *this, source, name, pn_link_name(link), node.relay));
        incoming[link] = in;
    } else {
        pn_terminus_set_type(pn_link_target(link), PN_UNSPECIFIED);
        throw Exception(qpid::amqp::error_conditions::NOT_FOUND, std::string("Node not found: ") + name);
    }
    if (connection.getBroker().getOptions().auth && !connection.isLink())
        incoming[link]->verify(connection.getUserId(), connection.getBroker().getOptions().realm);
    QPID_LOG(debug, "Incoming link attached");
}

void Session::setupOutgoing(pn_link_t* link, pn_terminus_t* source, const std::string& name)
{
    ResolvedNode node = resolve(name, source, false);
    if (node.queue) {
        setCapabilities(pn_terminus_capabilities(source), pn_terminus_capabilities(pn_link_source(link)), node.queue);
        node.properties.write(pn_terminus_properties(pn_link_source(link)), node.queue);
    } else if (node.exchange) {
        setCapabilities(pn_terminus_capabilities(source), pn_terminus_capabilities(pn_link_source(link)), node.exchange);
        node.properties.write(pn_terminus_properties(pn_link_source(link)), node.exchange);
    }

    Filter filter;
    filter.read(pn_terminus_filter(source));
    const char* targetAddress = pn_terminus_get_address(pn_link_remote_target(link));
    if (!targetAddress) {
        targetAddress = pn_terminus_get_address(pn_link_target(link));
    }
    std::string target;
    if (targetAddress) {
        target = targetAddress;
    }

    if (node.queue) {
        authorise.outgoing(node.queue);
        SubscriptionType type = pn_terminus_get_distribution_mode(source) == PN_DIST_MODE_COPY ? BROWSER : CONSUMER;
        boost::shared_ptr<Outgoing> q(new OutgoingFromQueue(connection.getBroker(), name, target, node.queue, link, *this, out, type, false, node.properties.trackControllingLink()));
        q->init();
        filter.apply(q);
        outgoing[link] = q;
    } else if (node.exchange) {
        authorise.access(node.exchange);//do separate access check before trying to create the queue
        bool shared = is_capability_requested(SHARED, pn_terminus_capabilities(source));
        bool durable = pn_terminus_get_durability(source);
        bool autodelete = !durable && pn_link_remote_snd_settle_mode(link) != PN_SND_UNSETTLED;
        QueueSettings settings(durable, autodelete);
        std::string altExchange;
        if (node.topic) {
            settings = node.topic->getPolicy();
            settings.durable = durable;
            settings.autodelete = autodelete;
            altExchange = node.topic->getAlternateExchange();
        }
        settings.autoDeleteDelay = pn_terminus_get_timeout(source);
        if (settings.autoDeleteDelay) {
            settings.autodelete = true;
            settings.original["qpid.auto_delete_timeout"] = settings.autoDeleteDelay;
        }
        filter.configure(settings);
        std::stringstream queueName;
        if (shared) {
            //just use link name (TODO: could allow this to be
            //overridden when access to link properties is provided
            //(PROTON-335))
            queueName << pn_link_name(link);
        } else {
            //combination of container id and link name is unique
            queueName << connection.getContainerId() << "_" << pn_link_name(link);
        }
        boost::shared_ptr<qpid::broker::Queue> queue
            = connection.getBroker().createQueue(queueName.str(), settings, this, altExchange, connection.getUserId(), connection.getId()).first;
        if (!shared) queue->setExclusiveOwner(this);
        authorise.outgoing(node.exchange, queue, filter);
        filter.bind(node.exchange, queue);
        boost::shared_ptr<Outgoing> q(new OutgoingFromQueue(connection.getBroker(), name, target, queue, link, *this, out, CONSUMER, !shared, false));
        q->init();
        outgoing[link] = q;
    } else if (node.relay) {
        boost::shared_ptr<Outgoing> out(new OutgoingFromRelay(link, connection.getBroker(), *this, name, target, pn_link_name(link), node.relay));
        outgoing[link] = out;
        out->init();
    } else {
        pn_terminus_set_type(pn_link_source(link), PN_UNSPECIFIED);
        throw Exception(qpid::amqp::error_conditions::NOT_FOUND, std::string("Node not found: ") + name);/*not-found*/
    }
    filter.write(pn_terminus_filter(pn_link_source(link)));
    QPID_LOG(debug, "Outgoing link attached");
}

/**
 * Called for links initiated by the broker
 */
void Session::attach(pn_link_t* link, const std::string& src, const std::string& tgt, boost::shared_ptr<Relay> relay)
{
    pn_terminus_t* source = pn_link_source(link);
    pn_terminus_t* target = pn_link_target(link);
    pn_terminus_set_address(source, src.c_str());
    pn_terminus_set_address(target, tgt.c_str());

    if (relay) {
        if (pn_link_is_sender(link)) {
            boost::shared_ptr<Outgoing> out(new OutgoingFromRelay(link, connection.getBroker(), *this, src, tgt, pn_link_name(link), relay));
            outgoing[link] = out;
            out->init();
        } else {
            boost::shared_ptr<Incoming> in(new IncomingToRelay(link, connection.getBroker(), *this, src, tgt, pn_link_name(link), relay));
            incoming[link] = in;
        }
    } else {
        if (pn_link_is_sender(link)) {
            setupOutgoing(link, source, src);
        } else {
            setupIncoming(link, target, tgt);
        }
    }
}

void Session::detach(pn_link_t* link)
{
    if (pn_link_is_sender(link)) {
        OutgoingLinks::iterator i = outgoing.find(link);
        if (i != outgoing.end()) {
            i->second->detached();
            boost::shared_ptr<Queue> q = OutgoingFromQueue::getExclusiveSubscriptionQueue(i->second.get());
            if (q) connection.getBroker().deleteQueue(q->getName(), connection.getUserId(), connection.getMgmtId());
            outgoing.erase(i);
            QPID_LOG(debug, "Outgoing link detached");
        }
    } else {
        IncomingLinks::iterator i = incoming.find(link);
        if (i != incoming.end()) {
            i->second->detached();
            incoming.erase(i);
            QPID_LOG(debug, "Incoming link detached");
        }
    }
}

void Session::accepted(pn_delivery_t* delivery, bool sync)
{
    if (sync) {
        //this is on IO thread
        pn_delivery_update(delivery, PN_ACCEPTED);
        pn_delivery_settle(delivery);//do we need to check settlement modes/orders?
        incomingMessageAccepted();
    } else {
        //this is not on IO thread, need to delay processing until on IO thread
        qpid::sys::Mutex::ScopedLock l(lock);
        if (!deleted) {
            completed.push_back(delivery);
            out.activateOutput();
        }
    }
}

void Session::readable(pn_link_t* link, pn_delivery_t* delivery)
{
    pn_delivery_tag_t tag = pn_delivery_tag(delivery);
    QPID_LOG(debug, "received delivery: " << std::string(tag.bytes, tag.size));
    incomingMessageReceived();
    IncomingLinks::iterator target = incoming.find(link);
    if (target == incoming.end()) {
        QPID_LOG(error, "Received message on unknown link");
        pn_delivery_update(delivery, PN_REJECTED);
        pn_delivery_settle(delivery);//do we need to check settlement modes/orders?
        incomingMessageRejected();
    } else {
        target->second->readable(delivery);
        if (target->second->haveWork()) out.activateOutput();
    }
}
void Session::writable(pn_link_t* link, pn_delivery_t* delivery)
{
    OutgoingLinks::iterator sender = outgoing.find(link);
    if (sender == outgoing.end()) {
        QPID_LOG(error, "Delivery returned for unknown link");
    } else {
        sender->second->handle(delivery);
    }
}

bool Session::dispatch()
{
    bool output(false);
    for (OutgoingLinks::iterator s = outgoing.begin(); s != outgoing.end();) {
        try {
            if (s->second->doWork()) output = true;
            ++s;
        } catch (const Exception& e) {
            pn_condition_t* error = pn_link_condition(s->first);
            pn_condition_set_name(error, e.symbol());
            pn_condition_set_description(error, e.what());
            pn_link_close(s->first);
            s->second->detached();
            outgoing.erase(s++);
            output = true;
        }
    }
    if (completed.size()) {
        output = true;
        std::deque<pn_delivery_t*> copy;
        {
            qpid::sys::Mutex::ScopedLock l(lock);
            completed.swap(copy);
        }
        for (std::deque<pn_delivery_t*>::iterator i = copy.begin(); i != copy.end(); ++i) {
            accepted(*i, true);
        }
    }
    for (IncomingLinks::iterator i = incoming.begin(); i != incoming.end();) {
        try {
            if (i->second->doWork()) output = true;
            ++i;
        } catch (const Exception& e) {
            pn_condition_t* error = pn_link_condition(i->first);
            pn_condition_set_name(error, e.symbol());
            pn_condition_set_description(error, e.what());
            pn_link_close(i->first);
            i->second->detached();
            incoming.erase(i++);
            output = true;
        }
    }

    return output;
}

void Session::close()
{
    for (OutgoingLinks::iterator i = outgoing.begin(); i != outgoing.end(); ++i) {
        i->second->detached();
    }
    for (IncomingLinks::iterator i = incoming.begin(); i != incoming.end(); ++i) {
        i->second->detached();
    }
    outgoing.clear();
    incoming.clear();
    QPID_LOG(debug, "Session " << session << " closed, all links detached.");
    for (std::set< boost::shared_ptr<Queue> >::const_iterator i = exclusiveQueues.begin(); i != exclusiveQueues.end(); ++i) {
        (*i)->releaseExclusiveOwnership();
    }
    exclusiveQueues.clear();
    qpid::sys::Mutex::ScopedLock l(lock);
    deleted = true;
}

void Session::wakeup()
{
    out.activateOutput();
}

Authorise& Session::getAuthorise()
{
    return authorise;
}

void IncomingToQueue::handle(qpid::broker::Message& message)
{
    if (queue->isDeleted()) {
        std::stringstream msg;
        msg << " Queue " << queue->getName() << " has been deleted";
        throw Exception(qpid::amqp::error_conditions::RESOURCE_DELETED, msg.str());
    }
    queue->deliver(message);
}

void IncomingToExchange::handle(qpid::broker::Message& message)
{
    authorise.route(exchange, message);
    DeliverableMessage deliverable(message, 0);
    exchange->route(deliverable);
    if (!deliverable.delivered) {
        if (exchange->getAlternate()) {
            exchange->getAlternate()->route(deliverable);
        }
    }
}

}}} // namespace qpid::broker::amqp