client.cc

Go to the documentation of this file.

// Copyright (C) 2018-2024 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/io_service_thread_pool.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/boost_time_utils.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;
using namespace boost::posix_time;
 
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(const IOServicePtr& 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);
 
    IOServicePtr io_service_;
 
    boost::weak_ptr<ConnectionPool> conn_pool_;
 
    Url url_;
 
    TlsContextPtr tls_context_;
 
    std::shared_ptr<TCPSocket<SocketCallback>> tcp_socket_;
 
    std::shared_ptr<TLSSocket<SocketCallback>> tls_socket_;
 
    IntervalTimerPtr 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(const IOServicePtr& 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:
        const size_t QUEUE_SIZE_THRESHOLD = 2048;
        const int QUEUE_WARN_SECS = 5;
 
        Destination(Url const& url, TlsContextPtr tls_context, size_t max_connections)
            : url_(url), tls_context_(tls_context),
              max_connections_(max_connections), connections_(), queue_(),
              last_queue_warn_time_(min_date_time), last_queue_size_(0) {
        }
 
        ~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 const& desc) {
            queue_.push(desc);
            size_t size = queue_.size();
            // If the queue size is larger than the threshold and growing, issue a
            // periodic warning.
            if ((size > QUEUE_SIZE_THRESHOLD) && (size > last_queue_size_)) {
                ptime now = microsec_clock::universal_time();
                if ((now - last_queue_warn_time_) > seconds(QUEUE_WARN_SECS)) {
                    LOG_WARN(http_logger, HTTP_CLIENT_QUEUE_SIZE_GROWING)
                             .arg(url_.toText())
                             .arg(size);
                    // Remember the last time we warned.
                    last_queue_warn_time_ = now;
                }
            }
 
            // Remember the previous size.
            last_queue_size_ = size;
        }
 
        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_;
 
        ptime last_queue_warn_time_;
 
        size_t last_queue_size_;
    };
 
    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);
        }
    }
 
    IOServicePtr io_service_;
 
    std::map<DestinationDescriptor, DestinationPtr> destinations_;
 
    std::mutex pool_mutex_;
 
    size_t max_url_connections_;
};
 
Connection::Connection(const IOServicePtr& io_service,
                       const TlsContextPtr& tls_context,
                       const ConnectionPoolPtr& conn_pool,
                       const Url& url)
    : io_service_(io_service), conn_pool_(conn_pool), url_(url),
      tls_context_(tls_context), tcp_socket_(), tls_socket_(),
      timer_(new IntervalTimer(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(const IOServicePtr& 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 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_));
 
            // Create the thread pool.
            thread_pool_.reset(new IoServiceThreadPool(thread_io_service_, thread_pool_size_,
                                                       defer_thread_start));
 
            LOG_DEBUG(http_logger, isc::log::DBGLVL_TRACE_BASIC, HTTP_CLIENT_MT_STARTED)
                     .arg(thread_pool_size_);
        } else {
            // Single-threaded mode: use the caller's IOService,
            // one connection per URL.
            conn_pool_.reset(new ConnectionPool(io_service, 1));
        }
    }
 
    ~HttpClientImpl() {
        stop();
    }
 
    void checkPermissions() {
        if (thread_pool_) {
            thread_pool_->checkPausePermissions();
        }
    }
 
    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();
        }
 
        if (thread_io_service_) {
            thread_io_service_->stopAndPoll();
            thread_io_service_->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_;
 
    IoServiceThreadPoolPtr thread_pool_;
};
 
HttpClient::HttpClient(const IOServicePtr& io_service, bool multi_threading_enabled,
                       size_t thread_pool_size, bool defer_thread_start) {
    if (!multi_threading_enabled && thread_pool_size) {
        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() {
    impl_->stop();
}
 
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::checkPermissions() {
    impl_->checkPermissions();
}
 
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