regulate_time_reached.hpp

/*
 * This file is part of libtcspc
 * Copyright 2019-2026 Board of Regents of the University of Wisconsin System
 * SPDX-License-Identifier: MIT
 */
 
#pragma once
 
#include "arg_wrappers.hpp"
#include "common.hpp"
#include "data_types.hpp"
#include "int_arith.hpp"
#include "introspect.hpp"
#include "processor.hpp"
#include "time_tagged_events.hpp"
 
#include <cstddef>
#include <limits>
#include <type_traits>
#include <utility>
 
namespace tcspc {
 
namespace internal {
 
template <typename DataTypes, typename Downstream>
class regulate_time_reached {
    static_assert(processor<Downstream, time_reached_event<DataTypes>>);
 
    using abstime_type = typename DataTypes::abstime_type;
 
    abstime_type interval_thresh;
    std::size_t count_thresh;
 
    abstime_type exact_reached = std::numeric_limits<abstime_type>::min();
    abstime_type next_time_thresh = std::numeric_limits<abstime_type>::min();
    std::size_t emitted_since_prev_time_reached = 0;
    std::size_t seen_since_prev_time_reached = 0;
 
    Downstream downstream;
 
    // Called for all upstream times seen.
    void handle_time_reached(abstime_type abstime) {
        ++seen_since_prev_time_reached;
        if (abstime >= next_time_thresh ||
            emitted_since_prev_time_reached >= count_thresh) {
            downstream.handle(time_reached_event<DataTypes>{abstime});
            next_time_thresh = add_sat(abstime, interval_thresh);
            emitted_since_prev_time_reached = 0;
            seen_since_prev_time_reached = 0;
        }
        exact_reached = abstime;
    }
 
  public:
    explicit regulate_time_reached(
        arg::interval_threshold<abstime_type> interval_threshold,
        arg::count_threshold<std::size_t> count_threshold,
        Downstream downstream)
        : interval_thresh(interval_threshold.value),
          count_thresh(count_threshold.value),
          downstream(std::move(downstream)) {}
 
    [[nodiscard]] auto introspect_node() const -> processor_info {
        return processor_info(this, "regulate_time_reached");
    }
 
    [[nodiscard]] auto introspect_graph() const -> processor_graph {
        return downstream.introspect_graph().push_entry_point(this);
    }
 
    template <typename DT> void handle(time_reached_event<DT> const &event) {
        static_assert(std::is_same_v<typename DT::abstime_type, abstime_type>);
        handle_time_reached(event.abstime);
    }
 
    // NOLINTNEXTLINE(cppcoreguidelines-rvalue-reference-param-not-moved)
    template <typename DT> void handle(time_reached_event<DT> &&event) {
        handle(static_cast<time_reached_event<DT> const &>(event));
    }
 
    template <typename OtherEvent>
        requires handler_for<Downstream, std::remove_cvref_t<OtherEvent>>
    void handle(OtherEvent &&event) {
        static_assert(std::is_same_v<decltype(event.abstime), abstime_type>);
        auto const abstime = event.abstime;
        downstream.handle(std::forward<OtherEvent>(event));
        ++emitted_since_prev_time_reached;
        handle_time_reached(abstime);
    }
 
    void flush() {
        // Emit time-reached for last seen event in order to convey the (best
        // known) stream end time on all downstream paths.
        // Only do so if we received at least one event and the last emitted
        // was something other than time-reached.
        if (exact_reached > std::numeric_limits<abstime_type>::min() &&
            seen_since_prev_time_reached > 0)
            downstream.handle(time_reached_event<DataTypes>{exact_reached});
        downstream.flush();
    }
};
 
} // namespace internal

template <typename DataTypes = default_data_types, typename Downstream>
auto regulate_time_reached(
    arg::interval_threshold<typename DataTypes::abstime_type>
        interval_threshold,
    arg::count_threshold<std::size_t> count_threshold, Downstream downstream) {
    return internal::regulate_time_reached<DataTypes, Downstream>(
        interval_threshold, count_threshold, std::move(downstream));
}
 
} // namespace tcspc