/*
 *
 * 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/AsyncResultHandle.h"
#include "qpid/broker/Broker.h"
#include "qpid/broker/ConfigAsyncContext.h"
#include "qpid/broker/DeliverableMessage.h"
#include "qpid/broker/Exchange.h"
#include "qpid/broker/ExchangeRegistry.h"
#include "qpid/broker/FedOps.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/amqp_0_10/MessageTransfer.h"
#include "qpid/framing/MessageProperties.h"
#include "qpid/framing/reply_exceptions.h"
#include "qpid/log/Statement.h"
#include "qpid/management/ManagementAgent.h"
#include "qpid/sys/ExceptionHolder.h"
#include <stdexcept>

namespace qpid {
namespace broker {

using std::string;

using namespace qpid::framing;
using qpid::framing::Buffer;
using qpid::framing::FieldTable;
using qpid::sys::Mutex;
using qpid::management::ManagementAgent;
using qpid::management::ManagementObject;
using qpid::management::Manageable;
using qpid::management::Args;
namespace _qmf = qmf::org::apache::qpid::broker;

namespace
{
    const std::string qpidMsgSequence("qpid.msg_sequence");
    const std::string qpidSequenceCounter("qpid.sequence_counter");
    const std::string qpidIVE("qpid.ive");
    const std::string QPID_MANAGEMENT("qpid.management");
}


Exchange::PreRoute::PreRoute(Deliverable& msg, Exchange* _p):parent(_p) {
    if (parent){
        if (parent->sequence || parent->ive) parent->sequenceLock.lock();

        if (parent->sequence){
            parent->sequenceNo++;
            msg.getMessage().addAnnotation(qpidMsgSequence,parent->sequenceNo);
        }
        if (parent->ive) {
            parent->lastMsg = msg.getMessage();
        }
    }
}

Exchange::PreRoute::~PreRoute(){
    if (parent && (parent->sequence || parent->ive)){
        parent->sequenceLock.unlock();
    }
}

namespace {
/** Store information about an exception to be thrown later.
 * If multiple exceptions are stored, save the first of the "most severe"
 * exceptions, SESSION is les sever than CONNECTION etc.
 */
class  ExInfo {
  public:
    enum Type { NONE, SESSION, CONNECTION, OTHER };

    ExInfo(string exchange) : type(NONE), exchange(exchange) {}
    void store(Type type_, const qpid::sys::ExceptionHolder& exception_, const boost::shared_ptr<Queue>& queue) {
        QPID_LOG(warning, "Exchange " << exchange << " cannot deliver to  queue "
                 <<  queue->getName() << ": " << exception_.what());
        if (type < type_) {     // Replace less severe exception
            type = type_;
            exception = exception_;
        }
    }

    void raise() {
        exception.raise();
    }

  private:
    Type type;
    string exchange;
    qpid::sys::ExceptionHolder exception;
};
}

void Exchange::doRoute(Deliverable& msg, ConstBindingList b)
{
    int count = 0;

    if (b.get()) {
        ExInfo error(getName()); // Save exception to throw at the end.
        for(std::vector<Binding::shared_ptr>::const_iterator i = b->begin(); i != b->end(); i++, count++) {
            try {
                msg.deliverTo((*i)->queue);
                if ((*i)->mgmtBinding != 0)
                    (*i)->mgmtBinding->inc_msgMatched();
            }
            catch (const SessionException& e) {
                error.store(ExInfo::SESSION, framing::createSessionException(e.code, e.what()),(*i)->queue);
            }
            catch (const ConnectionException& e) {
                error.store(ExInfo::CONNECTION, framing::createConnectionException(e.code, e.what()), (*i)->queue);
            }
            catch (const std::exception& e) {
                error.store(ExInfo::OTHER, qpid::sys::ExceptionHolder(new Exception(e.what())), (*i)->queue);
            }
        }
        error.raise();
    }

    if (mgmtExchange != 0)
    {
        qmf::org::apache::qpid::broker::Exchange::PerThreadStats *eStats = mgmtExchange->getStatistics();
        uint64_t contentSize = msg.contentSize();

        eStats->msgReceives += 1;
        eStats->byteReceives += contentSize;
        if (count == 0)
        {
            //QPID_LOG(warning, "Exchange " << getName() << " could not route message; no matching binding found");
            eStats->msgDrops += 1;
            eStats->byteDrops += contentSize;
            if (brokerMgmtObject)
                brokerMgmtObject->inc_discardsNoRoute();
        }
        else
        {
            eStats->msgRoutes += count;
            eStats->byteRoutes += count * contentSize;
        }
    }
}

void Exchange::routeIVE(){
    if (ive && lastMsg){
        DeliverableMessage dmsg(lastMsg, 0);
        route(dmsg);
    }
}


Exchange::Exchange (const string& _name, Manageable* parent, Broker* b) :
    name(_name), durable(false), alternateUsers(0), persistenceId(0), sequence(false),
    sequenceNo(0), ive(false), mgmtExchange(0), brokerMgmtObject(0), broker(b), destroyed(false)
{
    if (parent != 0 && broker != 0)
    {
        ManagementAgent* agent = broker->getManagementAgent();
        if (agent != 0)
        {
            mgmtExchange = new _qmf::Exchange (agent, this, parent, _name);
            mgmtExchange->set_durable(durable);
            mgmtExchange->set_autoDelete(false);
            agent->addObject(mgmtExchange, 0, durable);
            if (broker)
                brokerMgmtObject = (qmf::org::apache::qpid::broker::Broker*) broker->GetManagementObject();
        }
    }
}

Exchange::Exchange(const string& _name, bool _durable, const qpid::framing::FieldTable& _args,
                   Manageable* parent, Broker* b)
    : name(_name), durable(_durable), alternateUsers(0), persistenceId(0),
      args(_args), sequence(false), sequenceNo(0), ive(false), mgmtExchange(0), brokerMgmtObject(0), broker(b), destroyed(false)
{
    if (parent != 0 && broker != 0)
    {
        ManagementAgent* agent = broker->getManagementAgent();
        if (agent != 0)
        {
            mgmtExchange = new _qmf::Exchange (agent, this, parent, _name);
            mgmtExchange->set_durable(durable);
            mgmtExchange->set_autoDelete(false);
            mgmtExchange->set_arguments(ManagementAgent::toMap(args));
            agent->addObject(mgmtExchange, 0, durable);
            if (broker)
                brokerMgmtObject = (qmf::org::apache::qpid::broker::Broker*) broker->GetManagementObject();
        }
    }

    sequence = _args.get(qpidMsgSequence);
    if (sequence) {
        QPID_LOG(debug, "Configured exchange " <<  _name  << " with Msg sequencing");
        args.setInt64(std::string(qpidSequenceCounter), sequenceNo);
    }

    ive = _args.get(qpidIVE);
    if (ive) {
        if (broker && broker->isInCluster())
            throw framing::NotImplementedException("Cannot use Initial Value Exchanges in a cluster");
        QPID_LOG(debug, "Configured exchange " <<  _name  << " with Initial Value");
    }
}

Exchange::~Exchange ()
{
    if (mgmtExchange != 0)
        mgmtExchange->resourceDestroy ();
}

void Exchange::setAlternate(Exchange::shared_ptr _alternate)
{
    alternate = _alternate;
    if (mgmtExchange != 0) {
        if (alternate.get() != 0)
            mgmtExchange->set_altExchange(alternate->GetManagementObject()->getObjectId());
        else
            mgmtExchange->clr_altExchange();
    }
}

void Exchange::setPersistenceId(uint64_t id) const
{
    persistenceId = id;
}

Exchange::shared_ptr Exchange::decode(ExchangeRegistry& exchanges, Buffer& buffer)
{
    string name;
    string type;
    string altName;
    FieldTable args;

    buffer.getShortString(name);
    bool durable(buffer.getOctet());
    buffer.getShortString(type);
    buffer.get(args);
    // For backwards compatibility on restoring exchanges from before the alt-exchange update, perform check
    if (buffer.available())
        buffer.getShortString(altName);

    try {
        Exchange::shared_ptr exch = exchanges.declare(name, type, durable, args).first;
        exch->sequenceNo = args.getAsInt64(qpidSequenceCounter);
        exch->alternateName.assign(altName);
        return exch;
    } catch (const UnknownExchangeTypeException&) {
        QPID_LOG(warning, "Could not create exchange " << name << "; type " << type << " is not recognised");
        return Exchange::shared_ptr();
    }
}

void Exchange::encode(Buffer& buffer) const
{
    buffer.putShortString(name);
    buffer.putOctet(durable);
    buffer.putShortString(getType());
    if (args.isSet(qpidSequenceCounter))
        args.setInt64(std::string(qpidSequenceCounter),sequenceNo);
    buffer.put(args);
    buffer.putShortString(alternate.get() ? alternate->getName() : string(""));
}

uint32_t Exchange::encodedSize() const
{
    return name.size() + 1/*short string size*/
        + 1 /*durable*/
        + getType().size() + 1/*short string size*/
        + (alternate.get() ? alternate->getName().size() : 0) + 1/*short string size*/
        + args.encodedSize();
}

void Exchange::recoveryComplete(ExchangeRegistry& exchanges)
{
    if (!alternateName.empty()) {
        Exchange::shared_ptr ae = exchanges.find(alternateName);
        if (ae) setAlternate(ae);
        else QPID_LOG(warning, "Could not set alternate exchange \""
                      << alternateName << "\": does not exist.");
    }
}

ManagementObject* Exchange::GetManagementObject (void) const
{
    return (ManagementObject*) mgmtExchange;
}

void Exchange::registerDynamicBridge(DynamicBridge* db, AsyncStore* const store)
{
    if (!supportsDynamicBinding())
        throw Exception("Exchange type does not support dynamic binding");

    {
        Mutex::ScopedLock l(bridgeLock);
        for (std::vector<DynamicBridge*>::iterator iter = bridgeVector.begin();
             iter != bridgeVector.end(); iter++)
            (*iter)->sendReorigin();

        bridgeVector.push_back(db);
    }

    FieldTable args;
    args.setString(qpidFedOp, fedOpReorigin);
    bind(Queue::shared_ptr(), string(), &args, store);
}

void Exchange::removeDynamicBridge(DynamicBridge* db)
{
    Mutex::ScopedLock l(bridgeLock);
    for (std::vector<DynamicBridge*>::iterator iter = bridgeVector.begin();
         iter != bridgeVector.end(); iter++)
        if (*iter == db) {
            bridgeVector.erase(iter);
            break;
        }
}

void Exchange::handleHelloRequest()
{
}

void Exchange::propagateFedOp(const string& routingKey, const string& tags, const string& op, const string& origin, qpid::framing::FieldTable* extra_args)
{
    Mutex::ScopedLock l(bridgeLock);
    string myOp(op.empty() ? fedOpBind : op);

    for (std::vector<DynamicBridge*>::iterator iter = bridgeVector.begin();
         iter != bridgeVector.end(); iter++)
        (*iter)->propagateBinding(routingKey, tags, op, origin, extra_args);
}

Exchange::Binding::Binding(const string& _key, Queue::shared_ptr _queue, Exchange* _parent,
                           FieldTable _args, const string& _origin, ConfigHandle _cfgHandle)
    : parent(_parent), queue(_queue), key(_key), args(_args), origin(_origin), cfgHandle(_cfgHandle), mgmtBinding(0)
{
}

Exchange::Binding::~Binding ()
{
    if (mgmtBinding != 0) {
        ManagementObject* mo = queue->GetManagementObject();
        if (mo != 0)
            static_cast<_qmf::Queue*>(mo)->dec_bindingCount();
        mgmtBinding->resourceDestroy ();
    }
}

void Exchange::Binding::startManagement()
{
    if (parent != 0)
    {
        Broker* broker = parent->getBroker();
        if (broker != 0) {
            ManagementAgent* agent = broker->getManagementAgent();
            if (agent != 0) {
                ManagementObject* mo = queue->GetManagementObject();
                if (mo != 0) {
                    management::ObjectId queueId = mo->getObjectId();

                    mgmtBinding = new _qmf::Binding
                        (agent, this, (Manageable*) parent, queueId, key, ManagementAgent::toMap(args));
                    if (!origin.empty())
                        mgmtBinding->set_origin(origin);
                    agent->addObject(mgmtBinding);
                    static_cast<_qmf::Queue*>(mo)->inc_bindingCount();
                }
            }
        }
    }
}

ManagementObject* Exchange::Binding::GetManagementObject () const
{
    return (ManagementObject*) mgmtBinding;
}

uint64_t Exchange::Binding::getSize() { return 0; } // TODO: kpvdr: implement persistence
void Exchange::Binding::write(char* /*target*/) {} // TODO: kpvdr: implement persistence

void Exchange::persistBind(Binding::shared_ptr b, AsyncStore* const s) {
    if (s && broker != 0 && b->queue->isDurable() && isDurable()) {
        b->cfgHandle = s->createConfigHandle();
        boost::shared_ptr<BrokerAsyncContext> bc(new ConfigAsyncContext(&configureComplete, &broker->getAsyncResultQueue()));
        s->submitCreate(b->cfgHandle, b.get(), bc);
    }
}

void Exchange::persistUnbind(Binding::shared_ptr b, AsyncStore* const s) {
    if (s && broker != 0 && b->queue->isDurable() && isDurable()) {
        boost::shared_ptr<BrokerAsyncContext> bc(new ConfigAsyncContext(&configureComplete, &broker->getAsyncResultQueue()));
        s->submitDestroy(b->cfgHandle, bc);
        b->cfgHandle.reset();
    }
}

// static
void Exchange::configureComplete(const AsyncResultHandle* const arh) {
    std::cout << "@@@@ Exchange: Configure complete: err=" << arh->getErrNo() << "; msg=\"" << arh->getErrMsg() << "\"" << std::endl;
}

Exchange::MatchQueue::MatchQueue(Queue::shared_ptr q) : queue(q) {}

bool Exchange::MatchQueue::operator()(Exchange::Binding::shared_ptr b)
{
    return b->queue == queue;
}

//void Exchange::setProperties(Message& msg) {
//    qpid::broker::amqp_0_10::MessageTransfer::setExchange(msg, getName());
//}

bool Exchange::routeWithAlternate(Deliverable& msg)
{
    route(msg);
    if (!msg.delivered && alternate) {
        alternate->route(msg);
    }
    return msg.delivered;
}

}}