client_connection.cc

Go to the documentation of this file.

// Copyright (C) 2017-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/unix_domain_socket.h>
#include <cc/json_feed.h>
#include <config/client_connection.h>
#include <boost/enable_shared_from_this.hpp>
#include <array>
#include <functional>
 
using namespace isc::asiolink;
 
namespace isc {
namespace config {
 
class ClientConnectionImpl : public boost::enable_shared_from_this<ClientConnectionImpl> {
public:
 
    explicit ClientConnectionImpl(const IOServicePtr& io_service);
 
    void scheduleTimer(ClientConnection::Handler handler);
 
    void start(const ClientConnection::SocketPath& socket_path,
               const ClientConnection::ControlCommand& command,
               ClientConnection::Handler handler,
               const ClientConnection::Timeout& timeout);
 
    void stop();
 
    void doSend(const void* buffer, const size_t length,
                ClientConnection::Handler handler);
 
    void doReceive(ClientConnection::Handler handler);
 
    void terminate(const boost::system::error_code& ec,
                   ClientConnection::Handler handler);
 
    void timeoutCallback(ClientConnection::Handler handler);
 
private:
 
    UnixDomainSocket socket_;
 
    JSONFeedPtr feed_;
 
    std::string current_command_;
 
    std::array<char, 32768> read_buf_;
 
    IntervalTimer timer_;
 
    long timeout_;
};
 
ClientConnectionImpl::ClientConnectionImpl(const IOServicePtr& io_service)
    : socket_(io_service), feed_(), current_command_(), timer_(io_service),
      timeout_(0) {
}
 
void
ClientConnectionImpl::scheduleTimer(ClientConnection::Handler handler) {
    if (timeout_ > 0) {
        timer_.setup(std::bind(&ClientConnectionImpl::timeoutCallback,
                               this, handler),
                     timeout_, IntervalTimer::ONE_SHOT);
    }
}
 
void
ClientConnectionImpl::start(const ClientConnection::SocketPath& socket_path,
                            const ClientConnection::ControlCommand& command,
                            ClientConnection::Handler handler,
                            const ClientConnection::Timeout& timeout) {
    // Start the timer protecting against timeouts.
    timeout_ = timeout.timeout_;
    scheduleTimer(handler);
 
    // Store the command in the class member to make sure it is valid
    // the entire time.
    current_command_.assign(command.control_command_);
 
    // Pass self to lambda to make sure that the instance of this class
    // lives as long as the lambda is held for async connect.
    auto self(shared_from_this());
    // Start asynchronous connect. This will return immediately.
    socket_.asyncConnect(socket_path.socket_path_,
    [this, self, command, handler](const boost::system::error_code& ec) {
        // We failed to connect so we can't proceed. Simply clean up
        // and invoke the user callback to signal an error.
        if (ec) {
            // This doesn't throw.
            terminate(ec, handler);
 
        } else {
            // Connection successful. Transmit the command to the remote
            // endpoint asynchronously.
            doSend(current_command_.c_str(), current_command_.length(),
                   handler);
        }
    });
}
 
void
ClientConnectionImpl::doSend(const void* buffer, const size_t length,
                             ClientConnection::Handler handler) {
    // Pass self to lambda to make sure that the instance of this class
    // lives as long as the lambda is held for async send.
    auto self(shared_from_this());
    // Start asynchronous transmission of the command. This will return
    // immediately.
    socket_.asyncSend(buffer, length,
        [this, self, buffer, length, handler]
        (const boost::system::error_code& ec, size_t bytes_transferred) {
        // An error has occurred while sending. Close the connection and
        // signal an error.
        if (ec) {
            // This doesn't throw.
            terminate(ec, handler);
 
        } else {
            // Sending is in progress, so push back the timeout.
            scheduleTimer(handler);
 
            // If the number of bytes we have managed to send so far is
            // lower than the amount of data we're trying to send, we
            // have to schedule another send to deliver the rest of
            // the data.
            if (bytes_transferred < length) {
                doSend(static_cast<const char*>(buffer) + bytes_transferred,
                       length - bytes_transferred, handler);
 
            } else {
                // We have sent all the data. Start receiving a response.
                doReceive(handler);
            }
        }
    });
}
 
void
ClientConnectionImpl::doReceive(ClientConnection::Handler handler) {
    // Pass self to lambda to make sure that the instance of this class
    // lives as long as the lambda is held for async receive.
    auto self(shared_from_this());
    socket_.asyncReceive(&read_buf_[0], read_buf_.size(),
        [this, self, handler]
        (const boost::system::error_code& ec, size_t length) {
        // An error has occurred while receiving the data. Close the connection
        // and signal an error.
        if (ec) {
            // This doesn't throw.
            terminate(ec, handler);
 
        } else {
            // Receiving is in progress, so push back the timeout.
            scheduleTimer(handler);
 
            std::string x(&read_buf_[0], length);
            // Lazy initialization of the JSONFeed. The feed will be "parsing"
            // received JSON stream and will detect when the whole response
            // has been received.
            if (!feed_) {
                feed_.reset(new JSONFeed());
                feed_->initModel();
            }
            // Put everything we have received so far into the feed and process
            // the data.
            feed_->postBuffer(&read_buf_[0], length);
            feed_->poll();
            // If the feed indicates that only a part of the response has been
            // received, schedule another receive to get more data.
            if (feed_->needData()) {
                doReceive(handler);
 
            } else {
                // We have received the entire response, let's call the handler
                // and indicate success.
                terminate(ec, handler);
            }
        }
    });
}
 
void
ClientConnectionImpl::terminate(const boost::system::error_code& ec,
                                ClientConnection::Handler handler) {
    try {
        timer_.cancel();
        socket_.close();
        current_command_.clear();
        handler(ec, feed_);
 
    } catch (...) {
        // None of these operations should throw. In particular, the handler
        // should not throw but if it has been misimplemented, we want to make
        // sure we don't emit any exceptions from here.
    }
}
 
void
ClientConnectionImpl::timeoutCallback(ClientConnection::Handler handler) {
    // Timeout has occurred. The remote server didn't provide the entire
    // response within the given time frame. Let's close the connection
    // and signal the timeout.
    terminate(boost::asio::error::timed_out, handler);
}
 
ClientConnection::ClientConnection(const IOServicePtr& io_service)
    : impl_(new ClientConnectionImpl(io_service)) {
}
 
void
ClientConnection::start(const ClientConnection::SocketPath& socket_path,
                        const ClientConnection::ControlCommand& command,
                        ClientConnection::Handler handler,
                        const ClientConnection::Timeout& timeout) {
    impl_->start(socket_path, command, handler, timeout);
}
 
} // end of namespace config
} // end of namespace isc