Provides a mechanism to throttle the execution of actions on state updates

Throttler.cs

/// <summary>
/// Provides a mechanism to throttle the execution of actions on state updates.
/// </summary>
/// <typeparam name="T">The type of state to throttle. Must be a reference type.</typeparam>
public class Throttler<T> : IDisposable
    where T : class
{
    private readonly ILogger<Throttler<T>> _logger;
    private readonly IGrainBase _grain;
    private readonly TimeSpan _throttleInterval;
    private readonly Func<T?, Task> _stateAction;

    private T? _currentState;
    private IGrainTimer? _timer;
    private bool _hasUpdates;

    /// <summary>
    /// Initializes a new instance of the <see cref="Throttler{T}"/> class.
    /// </summary>
    /// <param name="logger">The logger used for logging information.</param>
    /// <param name="grain">The grain instance used to register timers.</param>
    /// <param name="stateAction">The action to execute when the state is processed.</param>
    /// <param name="throttleInterval">The time interval to throttle state updates. Defaults to 5 seconds if not specified.</param>
    /// <exception cref="ArgumentNullException">Thrown when <paramref name="logger"/> is null.</exception>
    public Throttler(ILogger<Throttler<T>> logger, IGrainBase grain, Func<T?, Task> stateAction, TimeSpan? throttleInterval = null)
    {
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _grain = grain;
        _throttleInterval = throttleInterval ?? TimeSpan.FromSeconds(5);
        _stateAction = stateAction;
    }

    /// <summary>
    /// Sends a state update to be processed according to throttling rules.
    /// </summary>
    /// <param name="state">The state to process.</param>
    /// <returns>A task representing the asynchronous operation.</returns>
    /// <remarks>
    /// If no throttling is currently active, the state will be processed immediately.
    /// Otherwise, the state will be stored and processed when the throttle timer elapses.
    /// </remarks>
    public async Task SendAsync(T? state)
    {
        _currentState = state;
        _hasUpdates = true;

        if (_timer is not null)
        {
            return;
        }

        // If the timer is null, it means we are not currently throttling
        // and we should log the state immediately.
        await ActOnState();
        StartTimer();
    }

    /// <summary>
    /// Disposes the resources used by the throttler.
    /// </summary>
    public void Dispose()
    {
        _timer?.Dispose();
        _timer = null;
    }

    /// <summary>
    /// Executes the state action with the current state.
    /// </summary>
    /// <returns>A task representing the asynchronous operation.</returns>
    private async Task ActOnState()
    {
        _logger.LogInformation("Processing state for the {Grain}", _grain.GrainContext.GrainId.ToString());
        await _stateAction(_currentState);
        _hasUpdates = false;
    }

    /// <summary>
    /// Starts the throttling timer with the specified interval.
    /// </summary>
    private void StartTimer() => _timer = _grain.RegisterGrainTimer(OnTimerElapsed, _throttleInterval, Timeout.InfiniteTimeSpan);

    /// <summary>
    /// Called when the throttling timer elapses to process any pending state updates.
    /// </summary>
    /// <returns>A task representing the asynchronous operation.</returns>
    private Task OnTimerElapsed()
    {
        if (_hasUpdates)
        {
            return ActOnState();
        }

        _timer?.Dispose();
        _timer = null;

        return Task.CompletedTask;
    }
}