I need help creating a sliding window rate limiter using Rx

Question

Can a rate limiter that can only handle a specified number for a specified time be implemented by using Rx?

For example, if you set it to process only 5 times per second and the input is as follows:

var input = new Queue<string>();

// I'll use System.Threading.Chaannels.Channel<T> here
input.Enqueue("data1"); // at 0.10s
input.Enqueue("data2"); // at 0.20s
input.Enqueue("data3"); // at 0.30s
input.Enqueue("data4"); // at 0.40s
input.Enqueue("data5"); // at 0.50s
input.Enqueue("data6"); // at 0.60s
input.Enqueue("data7"); // at 0.70s
input.Enqueue("data8"); // at 1.20s
input.Enqueue("data9"); // at 1.30s
input.Enqueue("data10"); // at 1.40s
input.Enqueue("data11"); // at 2.20s
input.Enqueue("data12"); // at 2.50s

// The rate limiter will watch the input queue and call OnNext() when it's time to process the next item
var rateLimiter = Observable.Empty<string>();

// Process the input data
rateLimiter.Subscribe(data => { });

// output:
// processing data1 at 0.10s
// processing data2 at 0.20s
// processing data3 at 0.30s
// processing data4 at 0.40s
// processing data5 at 0.50s
// processing data6 at 1.10s
// processing data7 at 1.20s
// processing data8 at 1.30s
// processing data9 at 1.40s
// processing data10 at 1.50s
// processing data11 at 2.20s
// processing data12 at 2.50s

Answer 1

If I understand your question correctly the following solution may suit you:

public static IObservable<T> Pace<T>(this IObservable<T> source, TimeSpan interval)
    => source
        .Select(p => Observable.Empty<T>().Delay(interval).StartWith(p))
        .Concat();

static void Main()
{
  var fastSequence = Observable
      .Interval(TimeSpan.FromMilliseconds(100))
      .Take(10);

  var rateLimitedSequence = fastSequence
      .Pace(TimeSpan.FromMilliseconds(150));

  rateLimitedSequence
      .TimeInterval()
      .Subscribe(i =>Console.WriteLine(i));


  Console.ReadLine();
}

Output:

0@00:00:00.1296188
1@00:00:00.1624077
2@00:00:00.1538291
3@00:00:00.1559773
4@00:00:00.1563906
5@00:00:00.1559143
6@00:00:00.1563431
7@00:00:00.1563380
8@00:00:00.1560298
9@00:00:00.1563190