BackgroundService.hpp

Go to the documentation of this file.

 
#pragma once
 
#include "corium/Service.hpp"
#include "corium/internal/VariantIndex.hpp"
 
#include <chrono>
#include <exception>
#include <stop_token>
#include <thread>
 
namespace corium {
 
template <
    typename EventVariantType = DefaultEvents,
    typename QueuePolicy = BoundedMpscQueuePolicy<EventVariantType, 1024>,
    typename SignalPolicy = CallbackSignalPolicy,
    typename StoragePolicy = DefaultStoragePolicy,
    typename OverflowPolicy = DropNewestOverflowPolicy
>
class BackgroundService : public Service<EventVariantType, QueuePolicy, SignalPolicy, StoragePolicy, OverflowPolicy> {
public:
    using Base = Service<EventVariantType, QueuePolicy, SignalPolicy, StoragePolicy, OverflowPolicy>;
    using EventVariant = typename Base::EventVariant;
 
    BackgroundService() = default;
 
    explicit BackgroundService(ServiceContextT<EventVariant> context)
        : Base(context)
    {}
 
    ~BackgroundService()
    {
        stop();
        join();
    }
 
    BackgroundService(const BackgroundService&) = delete;
    BackgroundService& operator=(const BackgroundService&) = delete;
 
    BackgroundService(BackgroundService&&) noexcept = default;
    BackgroundService& operator=(BackgroundService&&) noexcept = default;
 
    template <typename Rep, typename Period>
    std::size_t waitAndPump(const std::stop_token& stopToken, const std::chrono::duration<Rep, Period>& timeout)
    {
        if (this->incomingBus().empty() && !stopToken.stop_requested()) {
            this->incomingBus().signalPolicy().wait_for(timeout);
        }
 
        std::size_t processed = 0;
        while (!stopToken.stop_requested()) {
            if (!this->incomingBus().processOne()) {
                break;
            }
            processed++;
        }
        return processed;
    }
 
    template <typename Derived>
    void startThread(Derived* derived)
    {
        _thread = std::jthread([this, derived](std::stop_token stopToken) {
#if __cpp_exceptions
            using EvVariant = EventVariantType;
            try {
                derived->run(stopToken);
            } catch (const std::exception& e) {
                if constexpr (requires { derived->onError(std::current_exception()); }) {
                    derived->onError(std::current_exception());
                } else if constexpr (requires { derived->onError(e.what()); }) {
                    derived->onError(e.what());
                }
                if constexpr (has_variant_type_v<ErrorEvent, EvVariant>) {
                    this->postHighPriority(ErrorEvent{1, reinterpret_cast<uintptr_t>(e.what())});
                }
            } catch (...) {
                if constexpr (requires { derived->onError(std::current_exception()); }) {
                    derived->onError(std::current_exception());
                }
                if constexpr (has_variant_type_v<ErrorEvent, EvVariant>) {
                    this->postHighPriority(ErrorEvent{1, 0});
                }
            }
#else
            derived->run(stopToken);
#endif
        });
    }
 
    void stop() noexcept
    {
        if (_thread.joinable()) {
            _thread.request_stop();
        }
    }
 
    void join() noexcept
    {
        if (_thread.joinable()) {
            _thread.join();
        }
    }
 
private:
    std::jthread _thread;
};
 
template <typename EventVariant = DefaultEvents>
using ProducerBackgroundService = BackgroundService<
    EventVariant,
    NoQueuePolicy<EventVariant>,
    NoSignalPolicy,
    ZeroStoragePolicy
>;
 
template <typename EventVariant = DefaultEvents, std::size_t Capacity = 64>
using ConsumerBackgroundService = BackgroundService<
    EventVariant,
    BoundedMpscQueuePolicy<EventVariant, Capacity>,
    CallbackSignalPolicy,
    DefaultStoragePolicy
>;
 
} // namespace corium