job_builder.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include "job.h"
#include "retry_policy.h"
#include "cancellation_token.h"
 
#include <chrono>
#include <functional>
#include <memory>
#include <string>
#include <type_traits>
#include <vector>
 
namespace kcenon::thread
{
    class job_builder
    {
    public:
        job_builder() = default;
 
        auto name(const std::string& name) -> job_builder&
        {
            name_ = name;
            return *this;
        }
 
        auto work(std::function<common::VoidResult()> work_fn) -> job_builder&
        {
            work_fn_ = std::move(work_fn);
            return *this;
        }
 
        auto work_with_data(
            const std::vector<uint8_t>& data,
            std::function<common::VoidResult(const std::vector<uint8_t>&)> work_fn) -> job_builder&
        {
            data_ = data;
            data_work_fn_ = std::move(work_fn);
            return *this;
        }
 
        auto cancellation(const cancellation_token& token) -> job_builder&
        {
            cancellation_token_ = token;
            has_cancellation_ = true;
            return *this;
        }
 
        auto on_complete(std::function<void(common::VoidResult)> callback) -> job_builder&
        {
            on_complete_ = std::move(callback);
            return *this;
        }
 
        auto on_error(std::function<void(const common::error_info&)> callback) -> job_builder&
        {
            on_error_ = std::move(callback);
            return *this;
        }
 
        auto priority(job_priority prio) -> job_builder&
        {
            priority_ = prio;
            has_priority_ = true;
            return *this;
        }
 
        auto retry(const retry_policy& policy) -> job_builder&
        {
            retry_policy_ = policy;
            has_retry_ = true;
            return *this;
        }
 
        auto timeout(std::chrono::milliseconds timeout) -> job_builder&
        {
            timeout_ = timeout;
            has_timeout_ = true;
            return *this;
        }
 
        template<typename JobType, typename... Args>
        auto from(Args&&... args) -> job_builder&
        {
            static_assert(std::is_base_of_v<job, JobType>,
                "JobType must be derived from kcenon::thread::job");
 
            custom_job_factory_ = [args = std::make_tuple(std::forward<Args>(args)...)]() mutable
            {
                return std::apply([](auto&&... a) {
                    return std::make_unique<JobType>(std::forward<decltype(a)>(a)...);
                }, std::move(args));
            };
            use_custom_job_ = true;
            return *this;
        }
 
        [[nodiscard]] auto build() -> std::unique_ptr<job>
        {
            std::unique_ptr<job> result;
 
            if (use_custom_job_ && custom_job_factory_)
            {
                result = custom_job_factory_();
            }
            else
            {
                result = std::make_unique<built_job>(
                    name_.empty() ? "builder_job" : name_,
                    data_,
                    std::move(work_fn_),
                    std::move(data_work_fn_)
                );
            }
 
            // Apply composition
            if (has_cancellation_)
            {
                result->with_cancellation(cancellation_token_);
            }
            if (on_complete_)
            {
                result->with_on_complete(std::move(on_complete_));
            }
            if (on_error_)
            {
                result->with_on_error(std::move(on_error_));
            }
            if (has_priority_)
            {
                result->with_priority(priority_);
            }
            if (has_retry_)
            {
                result->with_retry(retry_policy_);
            }
            if (has_timeout_)
            {
                result->with_timeout(timeout_);
            }
 
            return result;
        }
 
        [[nodiscard]] auto build_shared() -> std::shared_ptr<job>
        {
            return build();
        }
 
    private:
        class built_job : public job
        {
        public:
            built_job(
                const std::string& name,
                const std::vector<uint8_t>& data,
                std::function<common::VoidResult()> work_fn,
                std::function<common::VoidResult(const std::vector<uint8_t>&)> data_work_fn)
                : job(data.empty() ? name : name)
                , work_fn_(std::move(work_fn))
                , data_work_fn_(std::move(data_work_fn))
            {
                if (!data.empty())
                {
                    data_ = data;
                }
            }
 
            [[nodiscard]] auto do_work() -> common::VoidResult override
            {
                // Check cancellation before starting
                if (cancellation_token_.is_cancelled())
                {
                    auto result = make_error_result(error_code::operation_canceled);
                    invoke_callbacks(result);
                    return result;
                }
 
                common::VoidResult result = [this]() -> common::VoidResult {
                    if (data_work_fn_ && !data_.empty())
                    {
                        return data_work_fn_(data_);
                    }
                    else if (work_fn_)
                    {
                        return work_fn_();
                    }
                    else
                    {
                        return common::error_info{
                            static_cast<int>(error_code::not_implemented),
                            "No work function provided to job_builder",
                            "thread_system"
                        };
                    }
                }();
 
                // Invoke callbacks
                invoke_callbacks(result);
 
                return result;
            }
 
        private:
            std::function<common::VoidResult()> work_fn_;
            std::function<common::VoidResult(const std::vector<uint8_t>&)> data_work_fn_;
        };
 
        // Builder state
        std::string name_;
        std::vector<uint8_t> data_;
        std::function<common::VoidResult()> work_fn_;
        std::function<common::VoidResult(const std::vector<uint8_t>&)> data_work_fn_;
 
        cancellation_token cancellation_token_;
        bool has_cancellation_{false};
 
        std::function<void(common::VoidResult)> on_complete_;
        std::function<void(const common::error_info&)> on_error_;
 
        job_priority priority_{job_priority::normal};
        bool has_priority_{false};
 
        retry_policy retry_policy_;
        bool has_retry_{false};
 
        std::chrono::milliseconds timeout_{0};
        bool has_timeout_{false};
 
        std::function<std::unique_ptr<job>()> custom_job_factory_;
        bool use_custom_job_{false};
    };
 
    [[nodiscard]] inline auto make_job() -> job_builder
    {
        return job_builder{};
    }
 
} // namespace kcenon::thread