path: root/src/lib/http/client.cc

// Copyright (C) 2018 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/tcp_socket.h>
#include <http/client.h>
#include <http/response_json.h>
#include <http/response_parser.h>
#include <boost/bind.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/weak_ptr.hpp>
#include <array>
#include <map>
#include <queue>

#include <iostream>

using namespace isc;
using namespace isc::asiolink;
using namespace http;

namespace {

/// @brief TCP socket callback function type.
typedef boost::function<void(boost::system::error_code ec, size_t length)>
SocketCallbackFunction;

/// @brief Socket callback class required by the TCPSocket API.
///
/// Its function call operator ignores callbacks invoked with "operation aborted"
/// error codes. Such status codes are generated when the posted IO operations
/// are canceled.
class SocketCallback {
public:

    /// @brief Constructor.
    ///
    /// Stores pointer to a callback function provided by a caller.
    ///
    //// @param socket_callback Pointer to a callback function.
    SocketCallback(SocketCallbackFunction socket_callback)
        : callback_(socket_callback) {
    }

    /// @brief Function call operator.
    ///
    /// Invokes the callback for all error codes except "operation aborted".
    ///
    /// @param ec Error code.
    /// @param length Length of the data transmitted over the socket.
    void operator()(boost::system::error_code ec, size_t length = 0) {
        if (ec.value() == boost::asio::error::operation_aborted) {
            return;
        }
        callback_(ec, length);
    }

private:

    /// @brief Holds pointer to a supplied callback.
    SocketCallbackFunction callback_;

};

class ConnectionPool;

/// @brief Shared pointer to a connection pool.
typedef boost::shared_ptr<ConnectionPool> ConnectionPoolPtr;

/// @brief Client side HTTP connection to the server.
///
/// Each connection is established with a unique destination identified by the
/// specified URL. Multiple requests to the same destination can be sent over
/// the same connection, if the connection is persistent. If the server closes
/// the TCP connection (e.g. after sending a response), the connection can
/// be re-established (using the same @c Connection object).
///
/// If new request is created while the previous request is still in progress,
/// the new request is stored in the FIFO queue. The queued requests to the
/// particular URL are sent to the server when the current transaction ends.
///
/// The communication over the TCP socket is asynchronous. The caller is notified
/// about the completion of the transaction via a callback that the caller supplies
/// when initiating the transaction.
class Connection : public boost::enable_shared_from_this<Connection> {
public:

    /// @brief Constructor.
    ///
    /// @param io_service IO service to be used for the connection.
    /// @param conn_pool Back pointer to the connection pool to which this connection
    /// belongs.
    /// @param url URL associated with this connection.
    explicit Connection(IOService& io_service, const ConnectionPoolPtr& conn_pool,
                        const Url& url);

    /// @brief Destructor.
    ~Connection();

    /// @brief Starts new asynchronous transaction (HTTP request and response).
    ///
    /// This method expects that all pointers provided as argument are non-null.
    ///
    /// @param request Pointer to the request to be sent to the server.
    /// @param response Pointer to the object into which the response is stored. The
    /// caller should create a response object of the type which matches the content
    /// type expected by the caller, e.g. HttpResponseJson when JSON content type
    /// is expected to be received.
    /// @param request_timeout Request timeout in milliseconds.
    /// @param callback Pointer to the callback function to be invoked when the
    /// transaction completes.
    void doTransaction(const HttpRequestPtr& request, const HttpResponsePtr& response,
                       const long request_timeout, const HttpClient::RequestHandler& callback);

    /// @brief Closes the socket and cancels the request timer.
    void close();

    /// @brief Checks if a transaction has been initiated over this connection.
    ///
    /// @return true if transaction has been initiated, false otherwise.
    bool isTransactionOngoing() const;

private:

    /// @brief Resets the state of the object.
    ///
    /// In particular, it removes instances of objects provided for the previous
    /// transaction by a caller. It doesn't close the socket, though.
    void resetState();

    /// @brief Performs tasks required after receiving a response or after an
    /// error.
    ///
    /// This method triggers user's callback, resets the state of the connection
    /// and initiates next transaction if there is any transaction queued for the
    /// URL associated with this connection.
    ///
    /// @param ec Error code received as a result of the IO operation.
    /// @param parsing_error Message parsing error.
    void terminate(const boost::system::error_code& ec,
                   const std::string& parsing_error = "");

    /// @brief Asynchronously sends data over the socket.
    ///
    /// The data sent over the socket are stored in the @c buf_.
    void doSend();

    /// @brief Asynchronously receives data over the socket.
    ///
    /// The data received over the socket are store into the @c input_buf_.
    void doReceive();

    /// @brief Local callback invoked when the connection is established.
    ///
    /// If the connection is successfully established, this callback will start
    /// to asynchronously send the request over the socket.
    ///
    /// @param request_timeout Request timeout specified for this transaction.
    /// @param ec Error code being a result of the connection attempt.
    void connectCallback(const long request_timeout,
                         const boost::system::error_code& ec);

    /// @brief Local callback invoked when an attempt to send a portion of data
    /// over the socket has ended.
    ///
    /// The portion of data that has been sent is removed from the buffer. If all
    /// data from the buffer were sent, the callback will start to asynchronously
    /// receive a response from the server.
    ///
    /// @param ec Error code being a result of sending the data.
    /// @param length Number of bytes sent.
    void sendCallback(const boost::system::error_code& ec, size_t length);

    /// @brief Local callback invoked when an attempt to receive a portion of data
    /// over the socket has ended.
    ///
    /// @param ec Error code being a result of receiving the data.
    /// @param length Number of bytes received.
    void receiveCallback(const boost::system::error_code& ec, size_t length);

    /// @brief Local callback invoked when request timeout occurs.
    void timerCallback();

    /// @brief Pointer to the connection pool owning this connection.
    ///
    /// This is a weak pointer to avoid circular dependency between the
    /// Connection and ConnectionPool.
    boost::weak_ptr<ConnectionPool> conn_pool_;

    /// @brief URL for this connection.
    Url url_;

    /// @brief Socket to be used for this connection.
    TCPSocket<SocketCallback> socket_;

    /// @brief Interval timer used for detecting request timeouts.
    IntervalTimer timer_;

    /// @brief Holds currently sent request.
    HttpRequestPtr current_request_;

    /// @brief Holds pointer to an object where response is to be stored.
    HttpResponsePtr current_response_;

    /// @brief Pointer to the HTTP response parser.
    HttpResponseParserPtr parser_;

    /// @brief User supplied callback.
    HttpClient::RequestHandler current_callback_;

    /// @brief Output buffer.
    std::string buf_;

    /// @brief Input buffer.
    std::array<char, 4096> input_buf_;
};

/// @brief Shared pointer to the connection.
typedef boost::shared_ptr<Connection> ConnectionPtr;

/// @brief Connection pool for managing multiple connections.
///
/// Connection pool creates and destroys connections. It holds pointers
/// to all created connections and can verify whether the particular
/// connection is currently busy or idle. If the connection is idle, it
/// uses this connection for new requests. If the connection is busy, it
/// queues new requests until the connection becomes available.
class ConnectionPool : public boost::enable_shared_from_this<ConnectionPool> {
public:

    /// @brief Constructor.
    ///
    /// @param io_service Reference to the IO service to be used by the
    /// connections.
    explicit ConnectionPool(IOService& io_service)
        : io_service_(io_service), conns_(), queue_() {
    }

    /// @brief Destructor.
    ///
    /// Closes all connections.
    ~ConnectionPool() {
        closeAll();
    }

    /// @brief Returns next queued request for the given URL.
    ///
    /// @param url URL for which next queued request should be retrieved.
    /// @param [out] request Pointer to the queued request.
    /// @param [out] response Pointer to the object into which response should
    /// be stored.
    /// @param request_timeout Requested timeout for the transaction.
    /// @param callback Pointer to the user callback for this request.
    ///
    /// @return true if the request for the given URL has been retrieved,
    /// false if there are no more requests queued for this URL.
    bool getNextRequest(const Url& url,
                        HttpRequestPtr& request,
                        HttpResponsePtr& response,
                        long& request_timeout,
                        HttpClient::RequestHandler& callback) {
        // Check if there is a queue for this URL. If there is no queue, there
        // is no request queued either.
        auto it = queue_.find(url);
        if (it != queue_.end()) {
            // If the queue is non empty, we take the oldest request.
            if (!it->second.empty()) {
                RequestDescriptor desc = it->second.front();
                it->second.pop();
                request = desc.request_;
                response = desc.response_;
                request_timeout = desc.request_timeout_,
                callback = desc.callback_;
                return (true);
            }
        }

        return (false);
    }

    /// @brief Queue next request for sending to the server.
    ///
    /// A new transaction is started immediatelly, if there is no other request
    /// in progress for the given URL. Otherwise, the request is queued.
    ///
    /// @param url Destination where the request should be sent.
    /// @param request Pointer to the request to be sent to the server.
    /// @param response Pointer to the object into which the response should be
    /// stored.
    /// @param request_timeout Requested timeout for the transaction in
    /// milliseconds.
    /// @param callback Pointer to the user callback to be invoked when the
    /// transaction ends.
    void queueRequest(const Url& url,
                      const HttpRequestPtr& request,
                      const HttpResponsePtr& response,
                      const long request_timeout,
                      const HttpClient::RequestHandler& callback) {
        auto it = conns_.find(url);
        if (it != conns_.end()) {
            ConnectionPtr conn = it->second;
            // There is a connection for this URL already. Check if it is idle.
            if (conn->isTransactionOngoing()) {
                // Connection is busy, so let's queue the request.
                queue_[url].push(RequestDescriptor(request, response,
                                                   request_timeout,
                                                   callback));

            } else {
                // Connection is idle, so we can start the transaction.
                conn->doTransaction(request, response, request_timeout,
                                    callback);
            }

        } else {
            // There is no connection with this destination yet. Let's create
            // it and start the transaction.
            ConnectionPtr conn(new Connection(io_service_, shared_from_this(),
                                              url));
            conn->doTransaction(request, response, request_timeout, callback);
            conns_[url] = conn;
        }
    }

    /// @brief Closes connection and removes associated information from the
    /// connection pool.
    ///
    /// @param url URL for which connection shuld be closed.
    void closeConnection(const Url& url) {
        // Close connection for the specified URL.
        auto conns_it = conns_.find(url);
        if (conns_it != conns_.end()) {
            conns_it->second->close();
            conns_.erase(conns_it);
        }

        // Remove requests from the queue.
        auto queue_it = queue_.find(url);
        if (queue_it != queue_.end()) {
            queue_.erase(queue_it);
        }
    }

    /// @brief Closes all connections and removes associated information from
    /// the connection pool.
    void closeAll() {
        for (auto conns_it = conns_.begin(); conns_it != conns_.end();
             ++conns_it) {
            conns_it->second->close();
        }

        conns_.clear();
        queue_.clear();
    }

private:

    /// @brief Holds reference to the IO service.
    IOService& io_service_;

    /// @brief Holds mapping of URLs to connections.
    std::map<Url, ConnectionPtr> conns_;

    /// @brief Request descriptor holds parameters associated with the
    /// particular request.
    struct RequestDescriptor {
        /// @brief Constructor.
        ///
        /// @param request Pointer to the request to be sent.
        /// @param response Pointer to the object into which the response will
        /// be stored.
        /// @param request_timeout Requested timeout for the transaction.
        /// @param callback Pointer to the user callback.
        RequestDescriptor(const HttpRequestPtr& request,
                          const HttpResponsePtr& response,
                          const long request_timeout,
                          const HttpClient::RequestHandler& callback)
            : request_(request), response_(response),
              request_timeout_(request_timeout),
              callback_(callback) {
        }

        /// @brief Holds pointer to the request.
        HttpRequestPtr request_;
        /// @brief Holds pointer to the response.
        HttpResponsePtr response_;
        /// @brief Holds requested timeout value.
        long request_timeout_;
        /// @brief Holds pointer to the user callback.
        HttpClient::RequestHandler callback_;
    };

    /// @brief Holds the queue of requests for different URLs.
    std::map<Url, std::queue<RequestDescriptor> > queue_;
};

Connection::Connection(IOService& io_service,
                       const ConnectionPoolPtr& conn_pool,
                       const Url& url)
    : conn_pool_(conn_pool), url_(url), socket_(io_service), timer_(io_service),
      current_request_(), current_response_(), parser_(), current_callback_(),
      buf_(), input_buf_() {
}

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

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

void
Connection::doTransaction(const HttpRequestPtr& request,
                          const HttpResponsePtr& response,
                          const long request_timeout,
                          const HttpClient::RequestHandler& callback) {
    try {
        current_request_ = request;
        current_response_ = response;
        parser_.reset(new HttpResponseParser(*current_response_));
        parser_->initModel();
        current_callback_ = callback;

        buf_ = request->toString();

        // 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 (socket_.getASIOSocket().is_open() && !socket_.isUsable()) {
            socket_.close();
        }

        /// @todo We're getting a hostname but in fact it is expected to be an IP address.
        /// We should extend the TCPEndpoint to also accept names. Currently, it will fall
        /// over for names.
        TCPEndpoint endpoint(url_.getStrippedHostname(),
                             static_cast<unsigned short>(url_.getPort()));
        SocketCallback socket_cb(boost::bind(&Connection::connectCallback, shared_from_this(),
                                             request_timeout, _1));

        // Establish new connection or use existing connection.
        socket_.open(&endpoint, socket_cb);

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

void
Connection::close() {
    timer_.cancel();
    socket_.close();
    resetState();
}

bool
Connection::isTransactionOngoing() const {
    return (static_cast<bool>(current_request_));
}

void
Connection::terminate(const boost::system::error_code& ec,
                      const std::string& parsing_error) {

    timer_.cancel();
    socket_.cancel();

    HttpResponsePtr response;

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

    try {
        // The callback should take care of its own exceptions but one
        // never knows.
        current_callback_(ec, response, parsing_error);

    } catch (...) {
    }

    // If we're not requesting connection persistence, we should close the socket.
    // We're going to reconnect for the next transaction.
    if (!current_request_->isPersistent()) {
        close();
    }

    resetState();

    // Check if there are any requests queued for this connection and start
    // another transaction if there is at least one.
    HttpRequestPtr request;
    long request_timeout;
    HttpClient::RequestHandler callback;
    ConnectionPoolPtr conn_pool = conn_pool_.lock();
    if (conn_pool && conn_pool->getNextRequest(url_, request, response, request_timeout,
                                               callback)) {
        doTransaction(request, response, request_timeout, callback);
    }
}

void
Connection::doSend() {
    SocketCallback socket_cb(boost::bind(&Connection::sendCallback, shared_from_this(),
                                         _1, _2));
    try {
        socket_.asyncSend(&buf_[0], buf_.size(), socket_cb);

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

void
Connection::doReceive() {
    TCPEndpoint endpoint;
    SocketCallback socket_cb(boost::bind(&Connection::receiveCallback, shared_from_this(),
                                         _1, _2));

    try {
        socket_.asyncReceive(static_cast<void*>(input_buf_.data()), input_buf_.size(), 0,
                             &endpoint, socket_cb);
    } catch (...) {
        terminate(boost::asio::error::not_connected);
    }
}

void
Connection::connectCallback(const long request_timeout, const boost::system::error_code& ec) {
    // 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.
    if (ec &&
        (ec.value() != boost::asio::error::in_progress) &&
        (ec.value() != boost::asio::error::already_connected)) {
        terminate(ec);

    } else {
        // Setup request timer.
        timer_.setup(boost::bind(&Connection::timerCallback, this), request_timeout,
                     IntervalTimer::ONE_SHOT);
        // Start sending the request asynchronously.
        doSend();
    }
}

void
Connection::sendCallback(const boost::system::error_code& ec, size_t length) {
    if (ec) {
        // EAGAIN and EWOULDBLOCK don't really indicate an error. The length
        // should be 0 in this case but let's be sure.
        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;
        }
    }

    // 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();

    } else {
        doSend();
    }
}

void
Connection::receiveCallback(const boost::system::error_code& ec, size_t length) {
    if (ec) {
        // 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;
        }
    }

    // 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()) {
        doReceive();

    } 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();
            terminate(ec);

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

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

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

}

namespace isc {
namespace http {

/// @brief HttpClient implementation.
class HttpClientImpl {
public:

    /// @brief Constructor.
    ///
    /// Creates new connection pool.
    HttpClientImpl(IOService& io_service)
        : conn_pool_(new ConnectionPool(io_service)) {
    }

    /// @brief Holds a pointer to the connection pool.
    ConnectionPoolPtr conn_pool_;

};

HttpClient::HttpClient(IOService& io_service)
    : impl_(new HttpClientImpl(io_service)) {
}

void
HttpClient::asyncSendRequest(const Url& url, const HttpRequestPtr& request,
                             const HttpResponsePtr& response,
                             const HttpClient::RequestHandler& callback,
                             const HttpClient::RequestTimeout& request_timeout) {
    if (!url.isValid()) {
        isc_throw(HttpClientError, "invalid URL specified for the HTTP client");
    }

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

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

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

    impl_->conn_pool_->queueRequest(url, request, response, request_timeout.value_,
                                    callback);
}

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

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