policy_queue_adapter.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <kcenon/thread/interfaces/pool_queue_adapter.h>
#include <kcenon/thread/policies/policy_queue.h>
#include <kcenon/thread/utils/formatter.h>
 
namespace kcenon::thread {
 
template<
    typename SyncPolicy,
    typename BoundPolicy = policies::unbounded_policy,
    typename OverflowPolicy = policies::overflow_reject_policy
>
class policy_queue_adapter : public pool_queue_adapter_interface {
public:
    using queue_type = policy_queue<SyncPolicy, BoundPolicy, OverflowPolicy>;
 
    policy_queue_adapter()
        : queue_(std::make_unique<queue_type>()) {}
 
    explicit policy_queue_adapter(BoundPolicy bound_policy)
        : queue_(std::make_unique<queue_type>(std::move(bound_policy))) {}
 
    policy_queue_adapter(SyncPolicy sync_policy,
                         BoundPolicy bound_policy,
                         OverflowPolicy overflow_policy)
        : queue_(std::make_unique<queue_type>(
            std::move(sync_policy),
            std::move(bound_policy),
            std::move(overflow_policy))) {}
 
    explicit policy_queue_adapter(std::unique_ptr<queue_type> queue)
        : queue_(std::move(queue)) {}
 
    ~policy_queue_adapter() override = default;
 
    // Non-copyable
    policy_queue_adapter(const policy_queue_adapter&) = delete;
    policy_queue_adapter& operator=(const policy_queue_adapter&) = delete;
 
    // Movable
    policy_queue_adapter(policy_queue_adapter&&) = default;
    policy_queue_adapter& operator=(policy_queue_adapter&&) = default;
 
    [[nodiscard]] auto enqueue(std::unique_ptr<job>&& j) -> common::VoidResult override {
        return queue_->enqueue(std::move(j));
    }
 
    [[nodiscard]] auto enqueue_batch(std::vector<std::unique_ptr<job>>&& jobs) -> common::VoidResult override {
        // policy_queue doesn't have enqueue_batch, so we implement it as a loop
        for (auto& j : jobs) {
            auto result = queue_->enqueue(std::move(j));
            if (result.is_err()) {
                return result;
            }
        }
        return common::ok();
    }
 
    [[nodiscard]] auto dequeue() -> common::Result<std::unique_ptr<job>> override {
        return queue_->dequeue();
    }
 
    [[nodiscard]] auto try_dequeue() -> common::Result<std::unique_ptr<job>> override {
        return queue_->try_dequeue();
    }
 
    [[nodiscard]] auto empty() const -> bool override {
        return queue_->empty();
    }
 
    [[nodiscard]] auto size() const -> std::size_t override {
        return queue_->size();
    }
 
    auto clear() -> void override {
        queue_->clear();
    }
 
    auto stop() -> void override {
        queue_->stop();
    }
 
    [[nodiscard]] auto is_stopped() const -> bool override {
        return queue_->is_stopped();
    }
 
    [[nodiscard]] auto get_capabilities() const -> queue_capabilities override {
        return queue_->get_capabilities();
    }
 
    [[nodiscard]] auto to_string() const -> std::string override {
        return utility_module::formatter::format(
            "policy_queue[size={}, stopped={}]",
            queue_->size(),
            queue_->is_stopped());
    }
 
    [[nodiscard]] auto get_job_queue() const -> std::shared_ptr<job_queue> override {
        // policy_queue is not a job_queue, return nullptr
        return nullptr;
    }
 
    [[nodiscard]] auto get_scheduler() -> scheduler_interface& override {
        return *queue_;
    }
 
    [[nodiscard]] auto get_scheduler() const -> const scheduler_interface& override {
        return *queue_;
    }
 
    [[nodiscard]] auto get_policy_queue() -> queue_type& {
        return *queue_;
    }
 
    [[nodiscard]] auto get_policy_queue() const -> const queue_type& {
        return *queue_;
    }
 
private:
    std::unique_ptr<queue_type> queue_;
};
 
// ============================================
// Type aliases for common configurations
// ============================================
 
using standard_queue_adapter = policy_queue_adapter<
    policies::mutex_sync_policy,
    policies::unbounded_policy,
    policies::overflow_reject_policy
>;
 
using lockfree_queue_adapter = policy_queue_adapter<
    policies::lockfree_sync_policy,
    policies::unbounded_policy,
    policies::overflow_reject_policy
>;
 
// ============================================
// Factory functions
// ============================================
 
template<typename PolicyQueueType>
[[nodiscard]] auto make_policy_queue_adapter()
    -> std::unique_ptr<pool_queue_adapter_interface>
{
    using adapter_type = policy_queue_adapter<
        typename PolicyQueueType::sync_policy_type,
        typename PolicyQueueType::bound_policy_type,
        typename PolicyQueueType::overflow_policy_type
    >;
    return std::make_unique<adapter_type>();
}
 
template<typename PolicyQueueType>
[[nodiscard]] auto make_policy_queue_adapter(
    typename PolicyQueueType::bound_policy_type bound_policy)
    -> std::unique_ptr<pool_queue_adapter_interface>
{
    using adapter_type = policy_queue_adapter<
        typename PolicyQueueType::sync_policy_type,
        typename PolicyQueueType::bound_policy_type,
        typename PolicyQueueType::overflow_policy_type
    >;
    return std::make_unique<adapter_type>(std::move(bound_policy));
}
 
[[nodiscard]] inline auto make_standard_queue_adapter()
    -> std::unique_ptr<pool_queue_adapter_interface>
{
    return std::make_unique<standard_queue_adapter>();
}
 
[[nodiscard]] inline auto make_lockfree_queue_adapter()
    -> std::unique_ptr<pool_queue_adapter_interface>
{
    return std::make_unique<lockfree_queue_adapter>();
}
 
} // namespace kcenon::thread