thread_pool_adapters.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <kcenon/network/detail/config/feature_flags.h>
#include "kcenon/network/integration/thread_integration.h"
 
#include <chrono>
#include <functional>
#include <future>
#include <memory>
#include <stdexcept>
#include <string>
 
#if KCENON_WITH_COMMON_SYSTEM
#include <kcenon/common/interfaces/executor_interface.h>
#include <kcenon/common/patterns/result.h>
#endif
 
namespace kcenon::network::integration {
 
#if KCENON_WITH_COMMON_SYSTEM
 
class function_job final : public ::kcenon::common::interfaces::IJob {
public:
    explicit function_job(std::function<void()> func, std::string name = "function_job")
        : func_(std::move(func)), name_(std::move(name)) {}
 
    ::kcenon::common::VoidResult execute() override {
        try {
            if (func_) {
                func_();
            }
            return ::kcenon::common::ok();
        } catch (const std::exception& ex) {
            return ::kcenon::common::VoidResult(::kcenon::common::error_info{
                ::kcenon::common::error_codes::INTERNAL_ERROR,
                ex.what(),
                "network_system::function_job"});
        } catch (...) {
            return ::kcenon::common::VoidResult(::kcenon::common::error_info{
                ::kcenon::common::error_codes::INTERNAL_ERROR,
                "Unknown exception in function_job",
                "network_system::function_job"});
        }
    }
 
    std::string get_name() const override { return name_; }
 
private:
    std::function<void()> func_;
    std::string name_;
};
 
class network_to_common_thread_adapter
    : public ::kcenon::common::interfaces::IExecutor {
public:
    [[nodiscard]] static ::kcenon::common::Result<std::shared_ptr<network_to_common_thread_adapter>>
    create(std::shared_ptr<thread_pool_interface> pool) {
        if (!pool) {
            return ::kcenon::common::Result<std::shared_ptr<network_to_common_thread_adapter>>::err(
                ::kcenon::common::error_codes::INVALID_ARGUMENT,
                "network_to_common_thread_adapter requires non-null pool",
                "network_to_common_thread_adapter::create");
        }
        return ::kcenon::common::Result<std::shared_ptr<network_to_common_thread_adapter>>::ok(
            std::shared_ptr<network_to_common_thread_adapter>(
                new network_to_common_thread_adapter(std::move(pool))));
    }
 
    ::kcenon::common::Result<std::future<void>> execute(
        std::unique_ptr<::kcenon::common::interfaces::IJob>&& job) override {
        if (!pool_ || !pool_->is_running()) {
            return ::kcenon::common::Result<std::future<void>>::err(
                ::kcenon::common::error_codes::INVALID_ARGUMENT,
                "Thread pool not running",
                "network_to_common_thread_adapter");
        }
 
        try {
            auto shared_job = std::shared_ptr<::kcenon::common::interfaces::IJob>(
                std::move(job));
            auto prom = std::make_shared<std::promise<void>>();
            auto fut = prom->get_future();
 
            pool_->submit([shared_job, prom]() mutable {
                auto result = shared_job->execute();
                if (result.is_err()) {
                    prom->set_exception(std::make_exception_ptr(
                        std::runtime_error(result.error().message)));
                } else {
                    prom->set_value();
                }
            });
 
            return ::kcenon::common::Result<std::future<void>>::ok(std::move(fut));
        } catch (const std::exception& e) {
            return ::kcenon::common::Result<std::future<void>>::err(
                ::kcenon::common::error_codes::INTERNAL_ERROR,
                e.what(),
                "network_to_common_thread_adapter");
        }
    }
 
    ::kcenon::common::Result<std::future<void>> execute_delayed(
        std::unique_ptr<::kcenon::common::interfaces::IJob>&& job,
        std::chrono::milliseconds delay) override {
        if (!pool_ || !pool_->is_running()) {
            return ::kcenon::common::Result<std::future<void>>::err(
                ::kcenon::common::error_codes::INVALID_ARGUMENT,
                "Thread pool not running",
                "network_to_common_thread_adapter");
        }
 
        try {
            auto shared_job = std::shared_ptr<::kcenon::common::interfaces::IJob>(
                std::move(job));
            auto prom = std::make_shared<std::promise<void>>();
            auto fut = prom->get_future();
 
            pool_->submit_delayed(
                [shared_job, prom]() mutable {
                    auto result = shared_job->execute();
                    if (result.is_err()) {
                        prom->set_exception(std::make_exception_ptr(
                            std::runtime_error(result.error().message)));
                    } else {
                        prom->set_value();
                    }
                },
                delay);
 
            return ::kcenon::common::Result<std::future<void>>::ok(std::move(fut));
        } catch (const std::exception& e) {
            return ::kcenon::common::Result<std::future<void>>::err(
                ::kcenon::common::error_codes::INTERNAL_ERROR,
                e.what(),
                "network_to_common_thread_adapter");
        }
    }
 
    size_t worker_count() const override {
        return pool_ ? pool_->worker_count() : 0;
    }
 
    bool is_running() const override {
        return pool_ ? pool_->is_running() : false;
    }
 
    size_t pending_tasks() const override {
        return pool_ ? pool_->pending_tasks() : 0;
    }
 
    void shutdown([[maybe_unused]] bool wait_for_completion = true) override {
        // thread_pool_interface doesn't expose shutdown
        // Lifecycle management should be done on the underlying pool directly
    }
 
private:
    explicit network_to_common_thread_adapter(
        std::shared_ptr<thread_pool_interface> pool)
        : pool_(std::move(pool)) {}
 
    std::shared_ptr<thread_pool_interface> pool_;
};
 
class common_to_network_thread_adapter : public thread_pool_interface {
public:
    [[nodiscard]] static ::kcenon::common::Result<std::shared_ptr<common_to_network_thread_adapter>>
    create(std::shared_ptr<::kcenon::common::interfaces::IExecutor> executor) {
        if (!executor) {
            return ::kcenon::common::Result<std::shared_ptr<common_to_network_thread_adapter>>::err(
                ::kcenon::common::error_codes::INVALID_ARGUMENT,
                "common_to_network_thread_adapter requires non-null executor",
                "common_to_network_thread_adapter::create");
        }
        return ::kcenon::common::Result<std::shared_ptr<common_to_network_thread_adapter>>::ok(
            std::shared_ptr<common_to_network_thread_adapter>(
                new common_to_network_thread_adapter(std::move(executor))));
    }
 
    std::future<void> submit(std::function<void()> task) override {
        if (!executor_ || !executor_->is_running()) {
            return make_error_future("Executor not running");
        }
 
        auto result = executor_->execute(
            std::make_unique<function_job>(std::move(task)));
 
        if (result.is_err()) {
            return make_error_future(result.error().message);
        }
 
        return std::move(result.value());
    }
 
    std::future<void> submit_delayed(
        std::function<void()> task,
        std::chrono::milliseconds delay) override {
        if (!executor_ || !executor_->is_running()) {
            return make_error_future("Executor not running");
        }
 
        auto result = executor_->execute_delayed(
            std::make_unique<function_job>(std::move(task)), delay);
 
        if (result.is_err()) {
            return make_error_future(result.error().message);
        }
 
        return std::move(result.value());
    }
 
    size_t worker_count() const override {
        return executor_ ? executor_->worker_count() : 0;
    }
 
    bool is_running() const override {
        return executor_ ? executor_->is_running() : false;
    }
 
    size_t pending_tasks() const override {
        return executor_ ? executor_->pending_tasks() : 0;
    }
 
    void shutdown(bool wait_for_completion = true) {
        if (executor_) {
            executor_->shutdown(wait_for_completion);
        }
    }
 
private:
    explicit common_to_network_thread_adapter(
        std::shared_ptr<::kcenon::common::interfaces::IExecutor> executor)
        : executor_(std::move(executor)) {}
 
    static std::future<void> make_error_future(const std::string& message) {
        std::promise<void> promise;
        promise.set_exception(std::make_exception_ptr(std::runtime_error(message)));
        return promise.get_future();
    }
 
    std::shared_ptr<::kcenon::common::interfaces::IExecutor> executor_;
};
 
#else // !KCENON_WITH_COMMON_SYSTEM
 
// Placeholder types when common_system is not available
struct function_job_unavailable final {
    function_job_unavailable() = delete;
};
 
struct network_to_common_thread_adapter_unavailable final {
    network_to_common_thread_adapter_unavailable() = delete;
};
 
struct common_to_network_thread_adapter_unavailable final {
    common_to_network_thread_adapter_unavailable() = delete;
};
 
#endif // KCENON_WITH_COMMON_SYSTEM
 
} // namespace kcenon::network::integration