client.cc

Go to the documentation of this file.

// Copyright (C) 2018-2021 Internet Systems Consortium, Inc. ("ISC")
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.

#include <config.h>

#include <asiolink/asio_wrapper.h>
#include <asiolink/interval_timer.h>
#include <asiolink/tls_socket.h>
#include <http/client.h>
#include <http/http_log.h>
#include <http/http_messages.h>
#include <http/response_json.h>
#include <http/response_parser.h>
#include <util/multi_threading_mgr.h>
#include <util/unlock_guard.h>

#include <boost/enable_shared_from_this.hpp>
#include <boost/weak_ptr.hpp>

#include <atomic>
#include <array>
#include <functional>
#include <iostream>
#include <map>
#include <mutex>
#include <queue>
#include <thread>


using namespace isc;
using namespace isc::asiolink;
using namespace isc::http;
using namespace isc::util;
namespace ph = std::placeholders;

namespace {

constexpr size_t MAX_LOGGED_MESSAGE_SIZE = 1024;

typedef std::function<void(boost::system::error_code ec, size_t length)>
SocketCallbackFunction;

class SocketCallback {
public:

    SocketCallback(SocketCallbackFunction socket_callback)
        : callback_(socket_callback) {
    }

    void operator()(boost::system::error_code ec, size_t length = 0) {
        if (ec.value() == boost::asio::error::operation_aborted) {
            return;
        }
        callback_(ec, length);
    }

private:

    SocketCallbackFunction callback_;

};

class ConnectionPool;

typedef boost::shared_ptr<ConnectionPool> ConnectionPoolPtr;

class Connection : public boost::enable_shared_from_this<Connection> {
public:

    explicit Connection(IOService& io_service,
                        const TlsContextPtr& tls_context,
                        const ConnectionPoolPtr& conn_pool,
                        const Url& url);

    ~Connection();

    void doTransaction(const HttpRequestPtr& request,
                       const HttpResponsePtr& response,
                       const long request_timeout,
                       const HttpClient::RequestHandler& callback,
                       const HttpClient::ConnectHandler& connect_callback,
                       const HttpClient::HandshakeHandler& handshake_callback,
                       const HttpClient::CloseHandler& close_callback);

    void close();

    bool isTransactionOngoing() const {
        return (started_);
    }

    bool isClosed() const {
        return (closed_);
    }

    void isClosedByPeer();

    bool isMySocket(int socket_fd) const;

    bool checkPrematureTimeout(const uint64_t transid);

private:

    void doTransactionInternal(const HttpRequestPtr& request,
                               const HttpResponsePtr& response,
                               const long request_timeout,
                               const HttpClient::RequestHandler& callback,
                               const HttpClient::ConnectHandler& connect_callback,
                               const HttpClient::HandshakeHandler& handshake_callback,
                               const HttpClient::CloseHandler& close_callback);

    void closeInternal();

    void isClosedByPeerInternal();

    bool checkPrematureTimeoutInternal(const uint64_t transid);

    void resetState();

    void terminate(const boost::system::error_code& ec,
                   const std::string& parsing_error = "");

    void terminateInternal(const boost::system::error_code& ec,
                           const std::string& parsing_error = "");

    bool runParser(const boost::system::error_code& ec, size_t length);

    bool runParserInternal(const boost::system::error_code& ec, size_t length);

    void scheduleTimer(const long request_timeout);

    void doHandshake(const uint64_t transid);

    void doSend(const uint64_t transid);

    void doReceive(const uint64_t transid);

    void connectCallback(HttpClient::ConnectHandler connect_callback,
                         const uint64_t transid,
                         const boost::system::error_code& ec);

    void handshakeCallback(HttpClient::HandshakeHandler handshake_callback,
                           const uint64_t transid,
                           const boost::system::error_code& ec);

    void sendCallback(const uint64_t transid, const boost::system::error_code& ec,
                      size_t length);

    void receiveCallback(const uint64_t transid, const boost::system::error_code& ec,
                         size_t length);

    void timerCallback();

    void closeCallback(const bool clear = false);

    boost::weak_ptr<ConnectionPool> conn_pool_;

    Url url_;

    TlsContextPtr tls_context_;

    std::unique_ptr<TCPSocket<SocketCallback> > tcp_socket_;

    std::unique_ptr<TLSSocket<SocketCallback> > tls_socket_;

    IntervalTimer timer_;

    HttpRequestPtr current_request_;

    HttpResponsePtr current_response_;

    HttpResponseParserPtr parser_;

    HttpClient::RequestHandler current_callback_;

    std::string buf_;

    std::array<char, 32768> input_buf_;

    uint64_t current_transid_;

    HttpClient::HandshakeHandler handshake_callback_;

    HttpClient::CloseHandler close_callback_;

    std::atomic<bool> started_;

    std::atomic<bool> need_handshake_;

    std::atomic<bool> closed_;

    std::mutex mutex_;
};

typedef boost::shared_ptr<Connection> ConnectionPtr;

class ConnectionPool : public boost::enable_shared_from_this<ConnectionPool> {
public:

    explicit ConnectionPool(IOService& io_service, size_t max_url_connections)
        : io_service_(io_service), destinations_(), pool_mutex_(),
          max_url_connections_(max_url_connections) {
    }

    ~ConnectionPool() {
        closeAll();
    }

    void processNextRequest(const Url& url, const TlsContextPtr& tls_context) {
        if (MultiThreadingMgr::instance().getMode()) {
            std::lock_guard<std::mutex> lk(pool_mutex_);
            return (processNextRequestInternal(url, tls_context));
        } else {
            return (processNextRequestInternal(url, tls_context));
        }
    }

    void postProcessNextRequest(const Url& url,
                                const TlsContextPtr& tls_context) {
        io_service_.post(std::bind(&ConnectionPool::processNextRequest,
                                   shared_from_this(), url, tls_context));
    }

    void queueRequest(const Url& url,
                      const TlsContextPtr& tls_context,
                      const HttpRequestPtr& request,
                      const HttpResponsePtr& response,
                      const long request_timeout,
                      const HttpClient::RequestHandler& request_callback,
                      const HttpClient::ConnectHandler& connect_callback,
                      const HttpClient::HandshakeHandler& handshake_callback,
                      const HttpClient::CloseHandler& close_callback) {
        if (MultiThreadingMgr::instance().getMode()) {
            std::lock_guard<std::mutex> lk(pool_mutex_);
            return (queueRequestInternal(url, tls_context, request, response,
                                         request_timeout, request_callback,
                                         connect_callback, handshake_callback,
                                         close_callback));
        } else {
            return (queueRequestInternal(url, tls_context, request, response,
                                         request_timeout, request_callback,
                                         connect_callback, handshake_callback,
                                         close_callback));
        }
    }

    void closeAll() {
        if (MultiThreadingMgr::instance().getMode()) {
            std::lock_guard<std::mutex> lk(pool_mutex_);
            closeAllInternal();
        } else {
            closeAllInternal();
        }
    }

    void closeIfOutOfBand(int socket_fd) {
        if (MultiThreadingMgr::instance().getMode()) {
            std::lock_guard<std::mutex> lk(pool_mutex_);
            closeIfOutOfBandInternal(socket_fd);
        } else {
            closeIfOutOfBandInternal(socket_fd);
        }
    }

private:

    void processNextRequestInternal(const Url& url,
                                    const TlsContextPtr& tls_context) {
        // Check if there is a queue for this URL. If there is no queue, there
        // is no request queued either.
        DestinationPtr destination = findDestination(url, tls_context);
        if (destination) {
            // Remove closed connections.
            destination->garbageCollectConnections();
            if (!destination->queueEmpty()) {
                // We have at least one queued request. Do we have an
                // idle connection?
                ConnectionPtr connection = destination->getIdleConnection();
                if (!connection) {
                    // No idle connections.
                    if (destination->connectionsFull()) {
                        return;
                    }
                    // Room to make another connection with this destination,
                    // so make one.
                    connection.reset(new Connection(io_service_, tls_context,
                                                    shared_from_this(), url));
                    destination->addConnection(connection);
                }

                // Dequeue the oldest request and start a transaction for it using
                // the idle connection.
                RequestDescriptor desc = destination->popNextRequest();
                connection->doTransaction(desc.request_, desc.response_,
                                          desc.request_timeout_, desc.callback_,
                                          desc.connect_callback_,
                                          desc.handshake_callback_,
                                          desc.close_callback_);
            }
        }
    }

    void queueRequestInternal(const Url& url,
                              const TlsContextPtr& tls_context,
                              const HttpRequestPtr& request,
                              const HttpResponsePtr& response,
                              const long request_timeout,
                              const HttpClient::RequestHandler& request_callback,
                              const HttpClient::ConnectHandler& connect_callback,
                              const HttpClient::HandshakeHandler& handshake_callback,
                              const HttpClient::CloseHandler& close_callback) {
        ConnectionPtr connection;
        // Find the destination for the requested URL.
        DestinationPtr destination = findDestination(url, tls_context);
        if (destination) {
            // Remove closed connections.
            destination->garbageCollectConnections();
            // Found it, look for an idle connection.
            connection = destination->getIdleConnection();
        } else {
            // Doesn't exist yet so it's a new destination.
            destination = addDestination(url, tls_context);
        }

        if (!connection) {
            if (destination->connectionsFull()) {
                // All connections busy, queue it.
                destination->pushRequest(RequestDescriptor(request, response,
                                                           request_timeout,
                                                           request_callback,
                                                           connect_callback,
                                                           handshake_callback,
                                                           close_callback));
                return;
            }

            // Room to make another connection with this destination, so make one.
            connection.reset(new Connection(io_service_, tls_context,
                                            shared_from_this(), url));
            destination->addConnection(connection);
        }

        // Use the connection to start the transaction.
        connection->doTransaction(request, response, request_timeout, request_callback,
                                  connect_callback, handshake_callback, close_callback);
    }

    void closeAllInternal() {
        for (auto const& destination : destinations_) {
            destination.second->closeAllConnections();
        }

        destinations_.clear();
    }

    void closeIfOutOfBandInternal(int socket_fd) {
        for (auto const& destination : destinations_) {
            // First we look for a connection with the socket.
            ConnectionPtr connection = destination.second->findBySocketFd(socket_fd);
            if (connection) {
                if (!connection->isTransactionOngoing()) {
                    // Socket has no transaction, so any ready event is
                    // out-of-band (other end probably closed), so
                    // let's close it.  Note we do not remove any queued
                    // requests, as this might somehow be occurring in
                    // between them.
                    destination.second->closeConnection(connection);
                }

                return;
            }
        }
    }

    struct RequestDescriptor {
        RequestDescriptor(const HttpRequestPtr& request,
                          const HttpResponsePtr& response,
                          const long& request_timeout,
                          const HttpClient::RequestHandler& callback,
                          const HttpClient::ConnectHandler& connect_callback,
                          const HttpClient::HandshakeHandler& handshake_callback,
                          const HttpClient::CloseHandler& close_callback)
            : request_(request), response_(response),
              request_timeout_(request_timeout), callback_(callback),
              connect_callback_(connect_callback),
              handshake_callback_(handshake_callback),
              close_callback_(close_callback) {
        }

        HttpRequestPtr request_;

        HttpResponsePtr response_;

        long request_timeout_;

        HttpClient::RequestHandler callback_;

        HttpClient::ConnectHandler connect_callback_;

        HttpClient::HandshakeHandler handshake_callback_;

        HttpClient::CloseHandler close_callback_;
    };

    typedef std::pair<Url, TlsContextPtr> DestinationDescriptor;

    class Destination {
    public:
        Destination(Url url, TlsContextPtr tls_context, size_t max_connections)
            : url_(url), tls_context_(tls_context),
              max_connections_(max_connections), connections_(), queue_() {
        }

        ~Destination() {
            closeAllConnections();
        }

        void addConnection(ConnectionPtr connection) {
            if (connectionsFull()) {
                isc_throw(BadValue, "URL: " << url_.toText()
                      << ", already at maximum connections: "
                      << max_connections_);
            }

            connections_.push_back(connection);
        }

        void closeConnection(ConnectionPtr connection) {
            for (auto it = connections_.begin(); it != connections_.end(); ++it) {
                if (*it == connection) {
                    (*it)->close();
                    connections_.erase(it);
                    break;
                }
            }
        }

        void closeAllConnections() {
            // Flush the queue.
            while (!queue_.empty()) {
                queue_.pop();
            }

            for (auto const& connection : connections_) {
                connection->close();
            }

            connections_.clear();
        }

        void garbageCollectConnections() {
            for (auto it = connections_.begin(); it != connections_.end();) {
                (*it)->isClosedByPeer();
                if (!(*it)->isClosed()) {
                    ++it;
                } else {
                    it = connections_.erase(it);
                }
            }
        }

        ConnectionPtr getIdleConnection() {
            for (auto const& connection : connections_) {
                if (!connection->isTransactionOngoing() &&
                    !connection->isClosed()) {
                    return (connection);
                }
            }

            return (ConnectionPtr());
        }

        ConnectionPtr findBySocketFd(int socket_fd) {
            for (auto const& connection : connections_) {
                if (connection->isMySocket(socket_fd)) {
                    return (connection);
                }
            }

            return (ConnectionPtr());
        }

        bool connectionsEmpty() {
            return (connections_.empty());
        }

        bool connectionsFull() {
            return (connections_.size() >= max_connections_);
        }

        size_t connectionCount() {
            return (connections_.size());
        }

        size_t getMaxConnections() const {
            return (max_connections_);
        }

        bool queueEmpty() const {
            return (queue_.empty());
        }

        void pushRequest(RequestDescriptor desc) {
            queue_.push(desc);
        }

        RequestDescriptor popNextRequest() {
            if (queue_.empty()) {
                isc_throw(InvalidOperation, "cannot pop, queue is empty");
            }

            RequestDescriptor desc = queue_.front();
            queue_.pop();
            return (desc);
        }

    private:
        Url url_;

        TlsContextPtr tls_context_;

        size_t max_connections_;

        std::list<ConnectionPtr> connections_;

        std::queue<RequestDescriptor> queue_;
    };

    typedef boost::shared_ptr<Destination> DestinationPtr;

    DestinationPtr addDestination(const Url& url,
                                  const TlsContextPtr& tls_context) {
        const DestinationDescriptor& desc = std::make_pair(url, tls_context);
        DestinationPtr destination(new Destination(url, tls_context,
                                                   max_url_connections_));
        destinations_[desc] = destination;
        return (destination);
    }

    DestinationPtr findDestination(const Url& url,
                                   const TlsContextPtr& tls_context) const {
        const DestinationDescriptor& desc = std::make_pair(url, tls_context);
        auto it = destinations_.find(desc);
        if (it != destinations_.end()) {
            return (it->second);
        }

        return (DestinationPtr());
    }

    void removeDestination(const Url& url,
                           const TlsContextPtr& tls_context) {
        const DestinationDescriptor& desc = std::make_pair(url, tls_context);
        auto it = destinations_.find(desc);
        if (it != destinations_.end()) {
            it->second->closeAllConnections();
            destinations_.erase(it);
        }
    }

    IOService& io_service_;

    std::map<DestinationDescriptor, DestinationPtr> destinations_;

    std::mutex pool_mutex_;

    size_t max_url_connections_;
};

Connection::Connection(IOService& io_service,
                       const TlsContextPtr& tls_context,
                       const ConnectionPoolPtr& conn_pool,
                       const Url& url)
    : conn_pool_(conn_pool), url_(url), tls_context_(tls_context),
      tcp_socket_(), tls_socket_(), timer_(io_service),
      current_request_(), current_response_(), parser_(),
      current_callback_(), buf_(), input_buf_(), current_transid_(0),
      close_callback_(), started_(false), need_handshake_(false),
      closed_(false) {
    if (!tls_context) {
        tcp_socket_.reset(new asiolink::TCPSocket<SocketCallback>(io_service));
    } else {
        tls_socket_.reset(new asiolink::TLSSocket<SocketCallback>(io_service,
                                                                  tls_context));
        need_handshake_ = true;
    }
}

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

void
Connection::resetState() {
    started_ = false;
    current_request_.reset();
    current_response_.reset();
    parser_.reset();
    current_callback_ = HttpClient::RequestHandler();
}

void
Connection::closeCallback(const bool clear) {
    if (close_callback_) {
        try {
            if (tcp_socket_) {
                close_callback_(tcp_socket_->getNative());
            } else if (tls_socket_) {
                close_callback_(tls_socket_->getNative());
            } else {
                isc_throw(Unexpected,
                          "internal error: can't find a socket to close");
            }
        } catch (...) {
            LOG_ERROR(http_logger, HTTP_CONNECTION_CLOSE_CALLBACK_FAILED);
        }
    }

    if (clear) {
        close_callback_ = HttpClient::CloseHandler();
    }
}

void
Connection::isClosedByPeer() {
    // This method applies only to idle connections.
    if (started_ || closed_) {
        return;
    }
    // This code was guarded by a lock so keep this.
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        isClosedByPeerInternal();
    } else {
        isClosedByPeerInternal();
    }
}

void
Connection::isClosedByPeerInternal() {
    // If the socket is open we check if it is possible to transmit
    // the data over this socket by reading from it with message
    // peeking. If the socket is not usable, we close it and then
    // re-open it. There is a narrow window of time between checking
    // the socket usability and actually transmitting the data over
    // this socket, when the peer may close the connection. In this
    // case we'll need to re-transmit but we don't handle it here.
    if (tcp_socket_) {
        if (tcp_socket_->getASIOSocket().is_open() &&
            !tcp_socket_->isUsable()) {
            closeCallback();
            closed_ = true;
            tcp_socket_->close();
        }
    } else if (tls_socket_) {
        if (tls_socket_->getASIOSocket().is_open() &&
            !tls_socket_->isUsable()) {
            closeCallback();
            closed_ = true;
            tls_socket_->close();
        }
    } else {
        isc_throw(Unexpected, "internal error: can't find the sending socket");
    }
}

void
Connection::doTransaction(const HttpRequestPtr& request,
                          const HttpResponsePtr& response,
                          const long request_timeout,
                          const HttpClient::RequestHandler& callback,
                          const HttpClient::ConnectHandler& connect_callback,
                          const HttpClient::HandshakeHandler& handshake_callback,
                          const HttpClient::CloseHandler& close_callback) {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        doTransactionInternal(request, response, request_timeout,
                              callback, connect_callback, handshake_callback,
                              close_callback);
    } else {
        doTransactionInternal(request, response, request_timeout,
                              callback, connect_callback, handshake_callback,
                              close_callback);
    }
}

void
Connection::doTransactionInternal(const HttpRequestPtr& request,
                                  const HttpResponsePtr& response,
                                  const long request_timeout,
                                  const HttpClient::RequestHandler& callback,
                                  const HttpClient::ConnectHandler& connect_callback,
                                  const HttpClient::HandshakeHandler& handshake_callback,
                                  const HttpClient::CloseHandler& close_callback) {
    try {
        started_ = true;
        current_request_ = request;
        current_response_ = response;
        parser_.reset(new HttpResponseParser(*current_response_));
        parser_->initModel();
        current_callback_ = callback;
        handshake_callback_ = handshake_callback;
        close_callback_ = close_callback;

        // Starting new transaction. Generate new transaction id.
        ++current_transid_;

        buf_ = request->toString();

        LOG_DEBUG(http_logger, isc::log::DBGLVL_TRACE_DETAIL,
                  HTTP_CLIENT_REQUEST_SEND)
            .arg(request->toBriefString())
            .arg(url_.toText());

        LOG_DEBUG(http_logger, isc::log::DBGLVL_TRACE_DETAIL_DATA,
                  HTTP_CLIENT_REQUEST_SEND_DETAILS)
            .arg(url_.toText())
            .arg(HttpMessageParserBase::logFormatHttpMessage(request->toString(),
                                                             MAX_LOGGED_MESSAGE_SIZE));

        // Setup request timer.
        scheduleTimer(request_timeout);

        TCPEndpoint endpoint(url_.getStrippedHostname(),
                             static_cast<unsigned short>(url_.getPort()));
        SocketCallback socket_cb(std::bind(&Connection::connectCallback, shared_from_this(),
                                           connect_callback, current_transid_,
                                           ph::_1));

        // Establish new connection or use existing connection.
        if (tcp_socket_) {
            tcp_socket_->open(&endpoint, socket_cb);
            return;
        }
        if (tls_socket_) {
            tls_socket_->open(&endpoint, socket_cb);
            return;
        }

        // Should never reach this point.
        isc_throw(Unexpected, "internal error: can't find a socket to open");

    } catch (const std::exception& ex) {
        // Re-throw with the expected exception type.
        isc_throw(HttpClientError, ex.what());
    }
}

void
Connection::close() {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        return (closeInternal());
    } else {
        return (closeInternal());
    }
}

void
Connection::closeInternal() {
    // Pass in true to discard the callback.
    closeCallback(true);

    closed_ = true;
    timer_.cancel();
    if (tcp_socket_) {
        tcp_socket_->close();
    }
    if (tls_socket_) {
        tls_socket_->close();
    }

    resetState();
}

bool
Connection::isMySocket(int socket_fd) const {
    if (tcp_socket_) {
        return (tcp_socket_->getNative() == socket_fd);
    } else if (tls_socket_) {
        return (tls_socket_->getNative() == socket_fd);
    }
    // Should never reach this point.
    std::cerr << "internal error: can't find my socket\n";
    return (false);
}

bool
Connection::checkPrematureTimeout(const uint64_t transid) {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        return (checkPrematureTimeoutInternal(transid));
    } else {
        return (checkPrematureTimeoutInternal(transid));
    }
}

bool
Connection::checkPrematureTimeoutInternal(const uint64_t transid) {
    // If there is no transaction but the handlers are invoked it means
    // that the last transaction in the queue timed out prematurely.
    // Also, if there is a transaction in progress but the ID of that
    // transaction doesn't match the one associated with the handler it,
    // also means that the transaction timed out prematurely.
    if (!isTransactionOngoing() || (transid != current_transid_)) {
        LOG_WARN(http_logger, HTTP_PREMATURE_CONNECTION_TIMEOUT_OCCURRED)
                .arg(isTransactionOngoing())
                .arg(transid)
                .arg(current_transid_);
        return (true);
    }

    return (false);
}

void
Connection::terminate(const boost::system::error_code& ec,
                      const std::string& parsing_error) {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        terminateInternal(ec, parsing_error);
    } else {
        terminateInternal(ec, parsing_error);
    }
}

void
Connection::terminateInternal(const boost::system::error_code& ec,
                              const std::string& parsing_error) {
    HttpResponsePtr response;
    if (isTransactionOngoing()) {

        timer_.cancel();
        if (tcp_socket_) {
            tcp_socket_->cancel();
        }
        if (tls_socket_) {
            tls_socket_->cancel();
        }

        if (!ec && current_response_->isFinalized()) {
            response = current_response_;

            LOG_DEBUG(http_logger, isc::log::DBGLVL_TRACE_BASIC,
                      HTTP_SERVER_RESPONSE_RECEIVED)
                .arg(url_.toText());

            LOG_DEBUG(http_logger, isc::log::DBGLVL_TRACE_BASIC_DATA,
                      HTTP_SERVER_RESPONSE_RECEIVED_DETAILS)
                .arg(url_.toText())
                .arg(parser_ ?
                     parser_->getBufferAsString(MAX_LOGGED_MESSAGE_SIZE) :
                     "[HttpResponseParser is null]");

        } else {
            std::string err = parsing_error.empty() ? ec.message() :
                                                      parsing_error;

            LOG_DEBUG(http_logger, isc::log::DBGLVL_TRACE_BASIC,
                      HTTP_BAD_SERVER_RESPONSE_RECEIVED)
                .arg(url_.toText())
                .arg(err);

            // Only log the details if we have received anything and tried
            // to parse it.
            if (!parsing_error.empty()) {
                LOG_DEBUG(http_logger, isc::log::DBGLVL_TRACE_BASIC_DATA,
                          HTTP_BAD_SERVER_RESPONSE_RECEIVED_DETAILS)
                    .arg(url_.toText())
                    .arg(parser_ ?
                         parser_->getBufferAsString(MAX_LOGGED_MESSAGE_SIZE) :
                         "[HttpResponseParser is null]");
            }
        }

        try {
            // The callback should take care of its own exceptions but one
            // never knows.
            if (MultiThreadingMgr::instance().getMode()) {
                UnlockGuard<std::mutex> lock(mutex_);
                current_callback_(ec, response, parsing_error);
            } else {
                current_callback_(ec, response, parsing_error);
            }
        } catch (...) {
        }

        // If we're not requesting connection persistence or the
        // connection has timed out, we should close the socket.
        if (!closed_ &&
            (!current_request_->isPersistent() ||
             (ec == boost::asio::error::timed_out))) {
            closeInternal();
        }

        resetState();
    }

    // Check if there are any requests queued for this destination and start
    // another transaction if there is at least one.
    ConnectionPoolPtr conn_pool = conn_pool_.lock();
    if (conn_pool) {
        conn_pool->postProcessNextRequest(url_, tls_context_);
    }
}

void
Connection::scheduleTimer(const long request_timeout) {
    if (request_timeout > 0) {
        timer_.setup(std::bind(&Connection::timerCallback, this), request_timeout,
                     IntervalTimer::ONE_SHOT);
    }
}

void
Connection::doHandshake(const uint64_t transid) {
    // Skip the handshake if it is not needed.
    if (!need_handshake_) {
        doSend(transid);
        return;
    }

    SocketCallback socket_cb(std::bind(&Connection::handshakeCallback,
                                       shared_from_this(),
                                       handshake_callback_,
                                       transid,
                                       ph::_1));
    try {
       tls_socket_->handshake(socket_cb);

    } catch (...) {
        terminate(boost::asio::error::not_connected);
    }
}

void
Connection::doSend(const uint64_t transid) {
    SocketCallback socket_cb(std::bind(&Connection::sendCallback,
                                       shared_from_this(),
                                       transid,
                                       ph::_1,
                                       ph::_2));
    try {
        if (tcp_socket_) {
            tcp_socket_->asyncSend(&buf_[0], buf_.size(), socket_cb);
            return;
        }

        if (tls_socket_) {
            tls_socket_->asyncSend(&buf_[0], buf_.size(), socket_cb);
            return;
        }

        // Should never reach this point.
        std::cerr << "internal error: can't find a socket to send to\n";
        isc_throw(Unexpected,
                  "internal error: can't find a socket to send to");
    } catch (...) {
        terminate(boost::asio::error::not_connected);
    }
}

void
Connection::doReceive(const uint64_t transid) {
    TCPEndpoint endpoint;
    SocketCallback socket_cb(std::bind(&Connection::receiveCallback,
                                       shared_from_this(),
                                       transid,
                                       ph::_1,
                                       ph::_2));
    try {
        if (tcp_socket_) {
            tcp_socket_->asyncReceive(static_cast<void*>(input_buf_.data()),
                                      input_buf_.size(), 0,
                                      &endpoint, socket_cb);
            return;
        }
        if (tls_socket_) {
            tls_socket_->asyncReceive(static_cast<void*>(input_buf_.data()),
                                      input_buf_.size(), 0,
                                      &endpoint, socket_cb);
            return;
        }
        // Should never reach this point.
        std::cerr << "internal error: can't find a socket to receive from\n";
        isc_throw(Unexpected,
                  "internal error: can't find a socket to receive from");

    } catch (...) {
        terminate(boost::asio::error::not_connected);
    }
}

void
Connection::connectCallback(HttpClient::ConnectHandler connect_callback,
                            const uint64_t transid,
                            const boost::system::error_code& ec) {
    if (checkPrematureTimeout(transid)) {
        return;
    }

    // Run user defined connect callback if specified.
    if (connect_callback) {
        // If the user defined callback indicates that the connection
        // should not be continued.
        if (tcp_socket_) {
            if (!connect_callback(ec, tcp_socket_->getNative())) {
                return;
            }
        } else if (tls_socket_) {
            if (!connect_callback(ec, tls_socket_->getNative())) {
                return;
            }
        } else {
            // Should never reach this point.
            std::cerr << "internal error: can't find a socket to connect\n";
        }
    }

    if (ec && (ec.value() == boost::asio::error::operation_aborted)) {
        return;

    // In some cases the "in progress" status code may be returned. It doesn't
    // indicate an error. Sending the request over the socket is expected to
    // be successful. Getting such status appears to be highly dependent on
    // the operating system.
    } else if (ec &&
        (ec.value() != boost::asio::error::in_progress) &&
        (ec.value() != boost::asio::error::already_connected)) {
        terminate(ec);

    } else {
        // Start the TLS handshake asynchronously.
        doHandshake(transid);
    }
}

void
Connection::handshakeCallback(HttpClient::ConnectHandler handshake_callback,
                              const uint64_t transid,
                              const boost::system::error_code& ec) {
    need_handshake_ = false;
    if (checkPrematureTimeout(transid)) {
        return;
    }

    // Run user defined handshake callback if specified.
    if (handshake_callback) {
        // If the user defined callback indicates that the connection
        // should not be continued.
        if (tls_socket_) {
            if (!handshake_callback(ec, tls_socket_->getNative())) {
                return;
            }
        } else {
            // Should never reach this point.
            std::cerr << "internal error: can't find TLS socket\n";
        }
    }

    if (ec && (ec.value() == boost::asio::error::operation_aborted)) {
        return;
    } else if (ec) {
        terminate(ec);

    } else {
        // Start sending the request asynchronously.
        doSend(transid);
    }
}

void
Connection::sendCallback(const uint64_t transid,
                         const boost::system::error_code& ec,
                         size_t length) {
    if (checkPrematureTimeout(transid)) {
        return;
    }

    if (ec) {
        if (ec.value() == boost::asio::error::operation_aborted) {
            return;

        // EAGAIN and EWOULDBLOCK don't really indicate an error. The length
        // should be 0 in this case but let's be sure.
        } else if ((ec.value() == boost::asio::error::would_block) ||
            (ec.value() == boost::asio::error::try_again)) {
            length = 0;

        } else {
            // Any other error should cause the transaction to terminate.
            terminate(ec);
            return;
        }
    }

    // Sending is in progress, so push back the timeout.
    scheduleTimer(timer_.getInterval());

    // If any data have been sent, remove it from the buffer and only leave the
    // portion that still has to be sent.
    if (length > 0) {
        buf_.erase(0, length);
    }

    // If there is no more data to be sent, start receiving a response. Otherwise,
    // continue sending.
    if (buf_.empty()) {
        doReceive(transid);

    } else {
        doSend(transid);
    }
}

void
Connection::receiveCallback(const uint64_t transid,
                            const boost::system::error_code& ec,
                            size_t length) {
    if (checkPrematureTimeout(transid)) {
        return;
    }

    if (ec) {
        if (ec.value() == boost::asio::error::operation_aborted) {
            return;

        // EAGAIN and EWOULDBLOCK don't indicate an error in this case. All
        // other errors should terminate the transaction.
        } if ((ec.value() != boost::asio::error::try_again) &&
              (ec.value() != boost::asio::error::would_block)) {
            terminate(ec);
            return;

        } else {
            // For EAGAIN and EWOULDBLOCK the length should be 0 anyway, but let's
            // make sure.
            length = 0;
        }
    }

    // Receiving is in progress, so push back the timeout.
    scheduleTimer(timer_.getInterval());

    if (runParser(ec, length)) {
        doReceive(transid);
    }
}

bool
Connection::runParser(const boost::system::error_code& ec, size_t length) {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        return (runParserInternal(ec, length));
    } else {
        return (runParserInternal(ec, length));
    }
}

bool
Connection::runParserInternal(const boost::system::error_code& ec,
                              size_t length) {
    // If we have received any data, let's feed the parser with it.
    if (length != 0) {
        parser_->postBuffer(static_cast<void*>(input_buf_.data()), length);
        parser_->poll();
    }

    // If the parser still needs data, let's schedule another receive.
    if (parser_->needData()) {
        return (true);

    } else if (parser_->httpParseOk()) {
        // No more data needed and parsing has been successful so far. Let's
        // try to finalize the response parsing.
        try {
            current_response_->finalize();
            terminateInternal(ec);

        } catch (const std::exception& ex) {
            // If there is an error here, we need to return the error message.
            terminateInternal(ec, ex.what());
        }

    } else {
        // Parsing was unsuccessful. Let's pass the error message held in the
        // parser.
        terminateInternal(ec, parser_->getErrorMessage());
    }

    return (false);
}

void
Connection::timerCallback() {
    // Request timeout occurred.
    terminate(boost::asio::error::timed_out);
}

}

namespace isc {
namespace http {

class HttpClientImpl {
public:
    HttpClientImpl(IOService& io_service, size_t thread_pool_size = 0,
                   bool defer_thread_start = false)
        : thread_pool_size_(thread_pool_size), thread_pool_() {
        if (thread_pool_size_ > 0) {
            // Create our own private IOService.
            thread_io_service_.reset(new IOService());

            // Create the thread pool.
            thread_pool_.reset(new HttpThreadPool(thread_io_service_, thread_pool_size_,
                                                  defer_thread_start));

            // Create the connection pool. Note that we use the thread_pool_size
            // as the maximum connections per URL value.
            conn_pool_.reset(new ConnectionPool(*thread_io_service_, thread_pool_size_));

            LOG_DEBUG(http_logger, isc::log::DBGLVL_TRACE_BASIC, HTTP_CLIENT_MT_STARTED)
                     .arg(getThreadCount());
        } else {
            // Single-threaded mode: use the caller's IOService,
            // one connection per URL.
            conn_pool_.reset(new ConnectionPool(io_service, 1));
        }
    }

    ~HttpClientImpl() {
        stop();
    }

    void start() {
        if (thread_pool_) {
            thread_pool_->run();
        }
    }

    void stop() {
        // Close all the connections.
        conn_pool_->closeAll();

        // Stop the thread pool.
        if (thread_pool_) {
            thread_pool_->stop();
        }
    }

    void pause() {
        if (!thread_pool_) {
            isc_throw(InvalidOperation, "HttpClient::pause - no thread pool");
        }

        // Pause the thread pool.
        thread_pool_->pause();
    }

    void resume() {
        if (!thread_pool_) {
            isc_throw(InvalidOperation, "HttpClient::resume - no thread pool");
        }

        // Resume running the thread pool.
        thread_pool_->run();
    }

    bool isRunning() {
        if (thread_pool_) {
            return (thread_pool_->isRunning());
        }

        return (false);
    }

    bool isStopped() {
        if (thread_pool_) {
            return (thread_pool_->isStopped());
        }

        return (false);
    }

    bool isPaused() {
        if (thread_pool_) {
            return (thread_pool_->isPaused());
        }

        return (false);
    }

    asiolink::IOServicePtr getThreadIOService() {
        return (thread_io_service_);
    };

    uint16_t getThreadPoolSize() {
        return (thread_pool_size_);
    }

    uint16_t getThreadCount() {
        if (!thread_pool_) {
            return (0);
        }
        return (thread_pool_->getThreadCount());
    }

    ConnectionPoolPtr conn_pool_;

private:

    size_t thread_pool_size_;

    asiolink::IOServicePtr thread_io_service_;

    HttpThreadPoolPtr thread_pool_;
};

HttpClient::HttpClient(IOService& io_service, size_t thread_pool_size,
                       bool defer_thread_start /* = false */) {
    if (thread_pool_size > 0) {
        if (!MultiThreadingMgr::instance().getMode()) {
            isc_throw(InvalidOperation,
                      "HttpClient thread_pool_size must be zero"
                      "when Kea core multi-threading is disabled");
        }
    }

    impl_.reset(new HttpClientImpl(io_service, thread_pool_size,
                                   defer_thread_start));
}

HttpClient::~HttpClient() {
}

void
HttpClient::asyncSendRequest(const Url& url,
                             const TlsContextPtr& tls_context,
                             const HttpRequestPtr& request,
                             const HttpResponsePtr& response,
                             const HttpClient::RequestHandler& request_callback,
                             const HttpClient::RequestTimeout& request_timeout,
                             const HttpClient::ConnectHandler& connect_callback,
                             const HttpClient::HandshakeHandler& handshake_callback,
                             const HttpClient::CloseHandler& close_callback) {
    if (!url.isValid()) {
        isc_throw(HttpClientError, "invalid URL specified for the HTTP client");
    }

    if ((url.getScheme() == Url::Scheme::HTTPS) && !tls_context) {
        isc_throw(HttpClientError, "HTTPS URL scheme but no TLS context");
    }

    if (!request) {
        isc_throw(HttpClientError, "HTTP request must not be null");
    }

    if (!response) {
        isc_throw(HttpClientError, "HTTP response must not be null");
    }

    if (!request_callback) {
        isc_throw(HttpClientError, "callback for HTTP transaction must not be null");
    }

    impl_->conn_pool_->queueRequest(url, tls_context, request, response,
                                    request_timeout.value_,
                                    request_callback, connect_callback,
                                    handshake_callback, close_callback);
}

void
HttpClient::closeIfOutOfBand(int socket_fd)  {
    return (impl_->conn_pool_->closeIfOutOfBand(socket_fd));
}

void
HttpClient::start() {
    impl_->start();
}

void
HttpClient::pause() {
    impl_->pause();
}

void
HttpClient::resume() {
    impl_->resume();
}

void
HttpClient::stop() {
    impl_->stop();
}

const IOServicePtr
HttpClient::getThreadIOService() const {
    return (impl_->getThreadIOService());
}

uint16_t
HttpClient::getThreadPoolSize() const {
    return (impl_->getThreadPoolSize());
}

uint16_t
HttpClient::getThreadCount() const {
    return (impl_->getThreadCount());
}

bool
HttpClient::isRunning() {
    return (impl_->isRunning());
}

bool
HttpClient::isStopped() {
    return (impl_->isStopped());
}

bool
HttpClient::isPaused() {
    return (impl_->isPaused());
}

} // end of namespace isc::http
} // end of namespace isc