// 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);

    }

}

​