conditions.hpp

#pragma once
 
#include "../../data_interfaces/source/ifdv2_condition.hpp"
 
#include <launchdarkly/async/cancellation.hpp>
#include <launchdarkly/async/promise.hpp>
 
#include <boost/asio/any_io_executor.hpp>
 
#include <chrono>
#include <cstdint>
#include <memory>
#include <mutex>
#include <optional>
#include <vector>
 
namespace launchdarkly::server_side::data_systems {
 
class TimedCondition : public data_interfaces::IFDv2Condition {
   public:
    TimedCondition(boost::asio::any_io_executor executor,
                   std::chrono::milliseconds timeout);
 
    ~TimedCondition() override;
 
    [[nodiscard]] async::Future<Type> Execute() override;
 
    void Close() override;
 
   protected:
    void ArmTimer();
 
    void CancelTimer();
 
   private:
    struct State {
        std::mutex mutex;
        bool closed = false;
        async::Promise<Type> promise;
        std::optional<async::CancellationSource> timer_cancel;
    };
 
    boost::asio::any_io_executor const executor_;
    std::chrono::milliseconds const timeout_;
    std::shared_ptr<State> const state_;
};
 
class FallbackCondition final : public TimedCondition {
   public:
    FallbackCondition(boost::asio::any_io_executor executor,
                      std::chrono::milliseconds timeout);
 
    void Inform(data_interfaces::FDv2SourceResult const& result) override;
 
    [[nodiscard]] Type GetType() const override { return Type::kFallback; }
};
 
class RecoveryCondition final : public TimedCondition {
   public:
    RecoveryCondition(boost::asio::any_io_executor executor,
                      std::chrono::milliseconds timeout);
 
    void Inform(data_interfaces::FDv2SourceResult const& result) override;
 
    [[nodiscard]] Type GetType() const override { return Type::kRecovery; }
};
 
class FallbackConditionFactory final
    : public data_interfaces::IFDv2ConditionFactory {
   public:
    FallbackConditionFactory(boost::asio::any_io_executor executor,
                             std::chrono::milliseconds timeout);
 
    [[nodiscard]] std::unique_ptr<data_interfaces::IFDv2Condition> Build()
        override;
 
    [[nodiscard]] data_interfaces::IFDv2Condition::Type GetType()
        const override;
 
   private:
    boost::asio::any_io_executor const executor_;
    std::chrono::milliseconds const timeout_;
};
 
class RecoveryConditionFactory final
    : public data_interfaces::IFDv2ConditionFactory {
   public:
    RecoveryConditionFactory(boost::asio::any_io_executor executor,
                             std::chrono::milliseconds timeout);
 
    [[nodiscard]] std::unique_ptr<data_interfaces::IFDv2Condition> Build()
        override;
 
    [[nodiscard]] data_interfaces::IFDv2Condition::Type GetType()
        const override;
 
   private:
    boost::asio::any_io_executor const executor_;
    std::chrono::milliseconds const timeout_;
};
 
class Conditions final {
   public:
    explicit Conditions(
        std::vector<std::unique_ptr<data_interfaces::IFDv2Condition>>
            conditions);
 
    ~Conditions();
 
    Conditions(Conditions const&) = delete;
    Conditions(Conditions&&) = delete;
    Conditions& operator=(Conditions const&) = delete;
    Conditions& operator=(Conditions&&) = delete;
 
    [[nodiscard]] async::Future<data_interfaces::IFDv2Condition::Type>
    GetFuture(async::CancellationToken token);
 
    void Inform(data_interfaces::FDv2SourceResult const& result);
 
    void Close();
 
   private:
    struct PendingEntry {
        std::int64_t id;
        async::Promise<data_interfaces::IFDv2Condition::Type> promise;
        std::unique_ptr<async::CancellationCallback> cancel_cb;
    };
 
    struct State {
        std::mutex mutex;
        std::int64_t next_id = 0;
        std::vector<PendingEntry> pending;
        std::optional<data_interfaces::IFDv2Condition::Type> aggregate_result;
    };
 
    std::vector<std::unique_ptr<data_interfaces::IFDv2Condition>> conditions_;
    std::shared_ptr<State> const state_;
};
 
}  // namespace launchdarkly::server_side::data_systems