/*
 *
 * 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/client/Connector.h"

#include "config.h"
#include "qpid/client/Bounds.h"
#include "qpid/client/ConnectionImpl.h"
#include "qpid/client/ConnectionSettings.h"
#include "qpid/Options.h"
#include "qpid/log/Statement.h"
#include "qpid/sys/Time.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/framing/InitiationHandler.h"
#include "qpid/sys/ssl/util.h"
#include "qpid/sys/AsynchIO.h"
#include "qpid/sys/ssl/SslSocket.h"
#include "qpid/sys/SocketAddress.h"
#include "qpid/sys/Dispatcher.h"
#include "qpid/sys/Poller.h"
#include "qpid/sys/SecuritySettings.h"
#include "qpid/Msg.h"

#include <iostream>
#include <boost/bind.hpp>
#include <boost/format.hpp>

namespace qpid {
namespace client {

using namespace qpid::sys;
using namespace qpid::sys::ssl;
using namespace qpid::framing;
using boost::format;
using boost::str;


class SslConnector : public Connector
{
    typedef std::deque<framing::AMQFrame> Frames;

    const uint16_t maxFrameSize;

    sys::Mutex lock;
    Frames frames;
    size_t lastEof; // Position after last EOF in frames
    uint64_t currentSize;
    Bounds* bounds;

    framing::ProtocolVersion version;
    bool initiated;
    bool closed;

    sys::ShutdownHandler* shutdownHandler;
    framing::InputHandler* input;

    sys::ssl::SslSocket socket;

    sys::AsynchConnector* connector;
    sys::AsynchIO* aio;
    std::string identifier;
    Poller::shared_ptr poller;
    SecuritySettings securitySettings;

    ~SslConnector();

    void readbuff(AsynchIO&, AsynchIOBufferBase*);
    void writebuff(AsynchIO&);
    void writeDataBlock(const framing::AMQDataBlock& data);
    void eof(AsynchIO&);
    void disconnected(AsynchIO&);

    void connect(const std::string& host, const std::string& port);
    void connected(const sys::Socket&);
    void connectFailed(const std::string& msg);

    void close();
    void send(framing::AMQFrame& frame);
    void abort();
    void connectAborted();

    void setInputHandler(framing::InputHandler* handler);
    void setShutdownHandler(sys::ShutdownHandler* handler);
    sys::ShutdownHandler* getShutdownHandler() const;
    framing::OutputHandler* getOutputHandler();
    const std::string& getIdentifier() const;
    const SecuritySettings* getSecuritySettings();
    void socketClosed(AsynchIO&, const Socket&);

    size_t decode(const char* buffer, size_t size);
    size_t encode(char* buffer, size_t size);
    bool canEncode();

public:
    SslConnector(Poller::shared_ptr p, framing::ProtocolVersion pVersion,
              const ConnectionSettings&,
              ConnectionImpl*);
};

// Static constructor which registers connector here
namespace {
    Connector* create(Poller::shared_ptr p, framing::ProtocolVersion v, const ConnectionSettings& s, ConnectionImpl* c) {
        return new SslConnector(p, v, s, c);
    }

    struct StaticInit {
        StaticInit() {
            try {
                CommonOptions common("", "", QPIDC_CONF_FILE);
                SslOptions options;
                common.parse(0, 0, common.clientConfig, true);
                options.parse (0, 0, common.clientConfig, true);
                if (options.certDbPath.empty()) {
                    QPID_LOG(info, "SSL connector not enabled, you must set QPID_SSL_CERT_DB to enable it.");
                } else {
                    initNSS(options);
                    Connector::registerFactory("ssl", &create);
                }
            } catch (const std::exception& e) {
                QPID_LOG(error, "Failed to initialise SSL connector: " << e.what());
            }
        };

        ~StaticInit() { shutdownNSS(); }
    } init;
}

SslConnector::SslConnector(Poller::shared_ptr p,
                     ProtocolVersion ver,
                     const ConnectionSettings& settings,
                     ConnectionImpl* cimpl)
    : maxFrameSize(settings.maxFrameSize),
      lastEof(0),
      currentSize(0),
      bounds(cimpl),
      version(ver),
      initiated(false),
      closed(true),
      shutdownHandler(0),
      input(0),
      aio(0),
      poller(p)
{
    QPID_LOG(debug, "SslConnector created for " << version.toString());

    if (settings.sslCertName != "") {
        QPID_LOG(debug, "ssl-cert-name = " << settings.sslCertName);
        socket.setCertName(settings.sslCertName);
    }
}

SslConnector::~SslConnector() {
    close();
}

void SslConnector::connect(const std::string& host, const std::string& port) {
    Mutex::ScopedLock l(lock);
    assert(closed);
    connector = AsynchConnector::create(
        socket,
        host, port,
        boost::bind(&SslConnector::connected, this, _1),
        boost::bind(&SslConnector::connectFailed, this, _3));
    closed = false;

    connector->start(poller);
}

void SslConnector::connected(const Socket&) {
    connector = 0;
    aio = AsynchIO::create(socket,
                           boost::bind(&SslConnector::readbuff, this, _1, _2),
                           boost::bind(&SslConnector::eof, this, _1),
                           boost::bind(&SslConnector::disconnected, this, _1),
                           boost::bind(&SslConnector::socketClosed, this, _1, _2),
                           0, // nobuffs
                           boost::bind(&SslConnector::writebuff, this, _1));

    aio->createBuffers(maxFrameSize);
    identifier = str(format("[%1%]") % socket.getFullAddress());
    ProtocolInitiation init(version);
    writeDataBlock(init);
    aio->start(poller);
}

void SslConnector::connectFailed(const std::string& msg) {
    connector = 0;
    QPID_LOG(warning, "Connect failed: " << msg);
    socket.close();
    if (!closed)
        closed = true;
    if (shutdownHandler)
        shutdownHandler->shutdown();
}

void SslConnector::close() {
    Mutex::ScopedLock l(lock);
    if (!closed) {
        closed = true;
        if (aio)
            aio->queueWriteClose();
    }
}

void SslConnector::socketClosed(AsynchIO&, const Socket&) {
    if (aio)
        aio->queueForDeletion();
    if (shutdownHandler)
        shutdownHandler->shutdown();
}

void SslConnector::connectAborted() {
    connector->stop();
    connectFailed("Connection timedout");
}

void SslConnector::abort() {
    // Can't abort a closed connection
    if (!closed) {
        if (aio) {
            // Established connection
            aio->requestCallback(boost::bind(&SslConnector::eof, this, _1));
        } else if (connector) {
            // We're still connecting
            connector->requestCallback(boost::bind(&SslConnector::connectAborted, this));
        }
    }
}

void SslConnector::setInputHandler(InputHandler* handler){
    input = handler;
}

void SslConnector::setShutdownHandler(ShutdownHandler* handler){
    shutdownHandler = handler;
}

OutputHandler* SslConnector::getOutputHandler() {
    return this;
}

sys::ShutdownHandler* SslConnector::getShutdownHandler() const {
    return shutdownHandler;
}

const std::string& SslConnector::getIdentifier() const {
    return identifier;
}

void SslConnector::send(AMQFrame& frame) {
    bool notifyWrite = false;
    {
    Mutex::ScopedLock l(lock);
    frames.push_back(frame);
    //only ask to write if this is the end of a frameset or if we
    //already have a buffers worth of data
    currentSize += frame.encodedSize();
    if (frame.getEof()) {
        lastEof = frames.size();
        notifyWrite = true;
    } else {
        notifyWrite = (currentSize >= maxFrameSize);
    }
    /*
      NOTE: Moving the following line into this mutex block
            is a workaround for BZ 570168, in which the test
            testConcurrentSenders causes a hang about 1.5%
            of the time.  ( To see the hang much more frequently
            leave this line out of the mutex block, and put a
            small usleep just before it.)

            TODO mgoulish - fix the underlying cause and then
                            move this call back outside the mutex.
    */
    if (notifyWrite && !closed) aio->notifyPendingWrite();
    }
}

void SslConnector::writebuff(AsynchIO& /*aio*/)
{
    // It's possible to be disconnected and be writable
    if (closed)
        return;

    if (!canEncode()) {
        return;
    }

    AsynchIOBufferBase* buffer = aio->getQueuedBuffer();
    if (buffer) {

        size_t encoded = encode(buffer->bytes, buffer->byteCount);

        buffer->dataStart = 0;
        buffer->dataCount = encoded;
        aio->queueWrite(buffer);
    }
}

// Called in IO thread.
bool SslConnector::canEncode()
{
    Mutex::ScopedLock l(lock);
    //have at least one full frameset or a whole buffers worth of data
    return lastEof || currentSize >= maxFrameSize;
}

// Called in IO thread.
size_t SslConnector::encode(char* buffer, size_t size)
{
    framing::Buffer out(buffer, size);
    size_t bytesWritten(0);
    {
        Mutex::ScopedLock l(lock);
        while (!frames.empty() && out.available() >= frames.front().encodedSize() ) {
            frames.front().encode(out);
            QPID_LOG(trace, "SENT [" << identifier << "]: " << frames.front());
            frames.pop_front();
            if (lastEof) --lastEof;
        }
        bytesWritten = size - out.available();
        currentSize -= bytesWritten;
    }
    if (bounds) bounds->reduce(bytesWritten);
    return bytesWritten;
}

void SslConnector::readbuff(AsynchIO& aio, AsynchIOBufferBase* buff)
{
    int32_t decoded = decode(buff->bytes+buff->dataStart, buff->dataCount);
    // TODO: unreading needs to go away, and when we can cope
    // with multiple sub-buffers in the general buffer scheme, it will
    if (decoded < buff->dataCount) {
        // Adjust buffer for used bytes and then "unread them"
        buff->dataStart += decoded;
        buff->dataCount -= decoded;
        aio.unread(buff);
    } else {
        // Give whole buffer back to aio subsystem
        aio.queueReadBuffer(buff);
    }
}

size_t SslConnector::decode(const char* buffer, size_t size)
{
    framing::Buffer in(const_cast<char*>(buffer), size);
    if (!initiated) {
        framing::ProtocolInitiation protocolInit;
        if (protocolInit.decode(in)) {
            QPID_LOG(debug, "RECV [" << identifier << "]: INIT(" << protocolInit << ")");
            if(!(protocolInit==version)){
                throw Exception(QPID_MSG("Unsupported version: " << protocolInit
                                         << " supported version " << version));
            }
        }
        initiated = true;
    }
    AMQFrame frame;
    while(frame.decode(in)){
        QPID_LOG(trace, "RECV [" << identifier << "]: " << frame);
        input->received(frame);
    }
    return size - in.available();
}

void SslConnector::writeDataBlock(const AMQDataBlock& data) {
    AsynchIOBufferBase* buff = aio->getQueuedBuffer();
    assert(buff);
    framing::Buffer out(buff->bytes, buff->byteCount);
    data.encode(out);
    buff->dataCount = data.encodedSize();
    aio->queueWrite(buff);
}

void SslConnector::eof(AsynchIO&) {
    close();
}

void SslConnector::disconnected(AsynchIO&) {
    close();
    socketClosed(*aio, socket);
}

const SecuritySettings* SslConnector::getSecuritySettings()
{
    securitySettings.ssf = socket.getKeyLen();
    securitySettings.authid = "dummy";//set to non-empty string to enable external authentication
    return &securitySettings;
}

}} // namespace qpid::client