Rx-Throttle

CSharp · Console

/Program.cs

// https://stackoverflow.com/a/69877166/1768303
#nullable enable
using System;
using System.Threading;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using static System.Console;

public class Program
{
    static async Task ThrottleWithScheduler()
    {
        WriteLine($"\n{nameof(ThrottleWithScheduler)}\n");

        var sc = new CustomSyncContext();
        var scheduler = new SynchronizationContextScheduler(sc);
        var subj = new BehaviorSubject<string>("A");

        subj
            .Do(v => WriteLine($"Emitted {v} on {sc.Elapsed}ms"))
            .Throttle(TimeSpan.FromMilliseconds(500), scheduler)
            .Subscribe(v => WriteLine($"Observed {v} on {sc.Elapsed}ms"));

        await Task.Delay(100);
        subj.OnNext("B");
        await Task.Delay(200);
        subj.OnNext("X");
        await Task.Delay(550);
        subj.OnNext("Y");

        await Task.Delay(2000);
        WriteLine("Finished!");
    }

    static async Task ThrottleWithObserveOn()
    {
        WriteLine($"\n{nameof(ThrottleWithObserveOn)}\n");

        var sc = new CustomSyncContext();
        var scheduler = new SynchronizationContextScheduler(sc);
        var subj = new BehaviorSubject<string>("A");

        subj
            .Do(v => WriteLine($"Emitted {v} on {sc.Elapsed}ms"))
            .Throttle(TimeSpan.FromMilliseconds(500))
            .ObserveOn(scheduler)
            .Subscribe(v => WriteLine($"Observed {v} on {sc.Elapsed}ms"));

        await Task.Delay(100);
        subj.OnNext("B");
        await Task.Delay(200);
        subj.OnNext("X");
        await Task.Delay(550);
        subj.OnNext("Y");

        await Task.Delay(2000);
        WriteLine("Finished!");
    }

    public static async Task Main()
    {
        await ThrottleWithScheduler();
        await ThrottleWithObserveOn();
    }
}

class CustomSyncContext : SynchronizationContext
{
    private readonly Stopwatch _sw = Stopwatch.StartNew();
    public long Elapsed { get { lock (_sw) { return _sw.ElapsedMilliseconds; } } }
    public override void Post(SendOrPostCallback d, object? state)
    {
        WriteLine($"Scheduled on {Elapsed}ms");
        Task.Delay(100).ContinueWith(
            continuationAction: _ =>
            {
                WriteLine($"Executed on {Elapsed}ms");
                d(state);
            },
            continuationOptions: TaskContinuationOptions.ExecuteSynchronously);
    }
}