client_connector.h

Go to the documentation of this file.

/*
 * Copyright 2023 MusicScience37 (Kenta Kabashima)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#pragma once
 
#include <atomic>
#include <functional>
#include <memory>
#include <mutex>
#include <utility>
#include <vector>
 
#include "msgpack_rpc/addresses/uri.h"
#include "msgpack_rpc/clients/impl/reconnection_timer.h"
#include "msgpack_rpc/common/status.h"
#include "msgpack_rpc/common/status_code.h"
#include "msgpack_rpc/config/reconnection_config.h"
#include "msgpack_rpc/executors/i_executor.h"
#include "msgpack_rpc/logging/logger.h"
#include "msgpack_rpc/transport/async_connect.h"
#include "msgpack_rpc/transport/backend_list.h"
#include "msgpack_rpc/transport/i_connection.h"
 
namespace msgpack_rpc::clients::impl {

class ClientConnector : public std::enable_shared_from_this<ClientConnector> {
public:
    using ConnectionCallback = std::function<void()>;

    using MessageReceivedCallback =
        transport::IConnection::MessageReceivedCallback;

    using MessageSentCallback = transport::IConnection::MessageSentCallback;

    using ConnectionClosedCallback = std::function<void()>;

    ClientConnector(const std::shared_ptr<executors::IExecutor>& executor,
        transport::BackendList backends,
        std::vector<addresses::URI> server_uris,
        const config::ReconnectionConfig& reconnection_config,
        const std::shared_ptr<logging::Logger>& logger)
        : backends_(std::move(backends)),
          server_uris_(std::move(server_uris)),
          retry_timer_(executor, logger, reconnection_config),
          logger_(logger) {}

    void start(ConnectionCallback on_connection,
        MessageReceivedCallback on_received, MessageSentCallback on_sent,
        ConnectionClosedCallback on_closed) {
        on_connection_ = std::move(on_connection);
        on_received_ = std::move(on_received);
        on_sent_ = std::move(on_sent);
        on_closed_ = std::move(on_closed);
        async_connect();
    }

    void stop() {
        if (is_stopped_.exchange(true, std::memory_order_acquire)) {
            return;
        }
 
        std::unique_lock<std::mutex> lock(connection_mutex_);
        if (connection_) {
            connection_->async_close();
            connection_.reset();
        }
        lock.unlock();
        retry_timer_.cancel();
    }

    [[nodiscard]] std::shared_ptr<transport::IConnection> connection() {
        std::unique_lock<std::mutex> lock(connection_mutex_);
        return connection_;
    }
 
private:
    void async_connect() {
        transport::async_connect(backends_, server_uris_,
            [self = this->shared_from_this()](const Status& status,
                const std::shared_ptr<transport::IConnection>& connection) {
                if (status.code() == StatusCode::SUCCESS) {
                    self->on_connection(connection);
                } else {
                    self->on_connection_failure();
                }
            });
    }

    void on_connection(
        const std::shared_ptr<transport::IConnection>& connection) {
        if (is_stopped_.load(std::memory_order_relaxed)) {
            return;
        }
        std::unique_lock<std::mutex> lock(connection_mutex_);
        connection_ = connection;
        connection->start(on_received_, on_sent_,
            [weak_self = weak_from_this()](const Status& /*status*/) {
                const auto self = weak_self.lock();
                if (self) {
                    self->on_connection_closed();
                }
            });
        retry_timer_.reset();
        lock.unlock();
 
        on_connection_();
    }

    void on_connection_failure() {
        if (is_stopped_.load(std::memory_order_relaxed)) {
            return;
        }
        retry_timer_.async_wait(
            [self = this->shared_from_this()] { self->async_connect(); });
    }

    void on_connection_closed() {
        if (is_stopped_.load(std::memory_order_relaxed)) {
            return;
        }
        std::unique_lock<std::mutex> lock(connection_mutex_);
        connection_.reset();
        lock.unlock();
        MSGPACK_RPC_TRACE(logger_, "Connection closed, so reconnect now.");
        on_closed_();
        async_connect();
    }

    transport::BackendList backends_;

    std::vector<addresses::URI> server_uris_;

    std::shared_ptr<transport::IConnection> connection_{};

    std::mutex connection_mutex_{};

    std::atomic<bool> is_stopped_{false};

    ReconnectionTimer retry_timer_;

    ConnectionCallback on_connection_{};

    MessageReceivedCallback on_received_{};

    MessageSentCallback on_sent_{};

    ConnectionClosedCallback on_closed_{};

    std::shared_ptr<logging::Logger> logger_;
};
 
}  // namespace msgpack_rpc::clients::impl