How to build a rate-limiting API with Observables?

Question

I would like to create a simple Calculator service that has a single method to add numbers. This Add method should be async and has to limit the number of concurrent calls being made at a given time. For instance, no more than 5 concurrent calls per second. If the rate limit is exceeded, the call should throw an exception.

The class should be like:

public class RateLimitingCalculator
{
    public async Task<int> Add(int a, int b) 
    {
        //...
    }
}

Any ideas? I would like implement it with Reactive Extensions, but if it's better to use another strategy, I would stick to it!

there are some ideas in [this post](http://stackoverflow.com/a/36933908/3407841) that may help you. — leetibbett, Aug 05 '16 at 11:29
You should really avoid mixing monads. I think it would be easier to reason this out if you had this signature: `public IObservable Add(IObservable values)`. — Enigmativity, Aug 05 '16 at 15:01
@Enigmativity What do you mean with "avoid mixing monads"? Also, do you think it makes sense to use Rx to rate-limit an API? — SuperJMN, Aug 06 '16 at 13:42
@SuperJMN - `IObservable` and `Task` are both monads. They are "super-types" taking the underlying `T` and giving it a super power. If you stay in your monad then everything works fine, but if you go between two monads then you can get errors. So I would avoid trying to use one monad (Rx) to rate-limit another (TPL). Stay with Rx, it has all the operators you need for rate limiting. — Enigmativity, Aug 07 '16 at 02:13
*no more than 5 concurrent calls per second* <--- What does this means? Is it OK to have 100 calls per second, but none of them concurrent with each other? — Theodor Zoulias, May 16 '20 at 23:17

score 1 · Accepted Answer · answered Aug 05 '16 at 18:32

I don't think using Rx makes sense here, unless you can rewrite your method into something like public IObservable<int> Add(IObservable<Tuple<int, int>> values), as suggested by Enigmativity in a comment.

What I would do is to separate the concern of rate limiting into a separate class. That way, your code could look something like this:

public class RateLimitingCalculator
{
    private RateLimiter rateLimiter = new RateLimiter(5, TimeSpan.FromSeconds(1));

    public async Task<int> Add(int a, int b) 
    {
        rateLimiter.ThrowIfRateExceeded();

        //...
    }
}

The implementation of RateLimiter depends on your exact requirements, but a very simple, not-thread-safe version could look like this:

class RateLimiter
{
    private readonly int rate;
    private readonly TimeSpan perTime;

    private DateTime secondStart = DateTime.MinValue;
    private int count = 0;

    public RateLimiter(int rate, TimeSpan perTime)
    {
        this.rate = rate;
        this.perTime = perTime;
    }

    public void ThrowIfRateExceeded()
    {
        var now = DateTime.UtcNow;

        if (now - secondStart > perTime)
        {
            secondStart = now;
            count = 1;
            return;
        }

        if (count >= rate)
            throw new RateLimitExceededException();

        count++;
    }
}

I also want do the opposite consuming a rate-limiing API as transparently as possible: http://stackoverflow.com/questions/38675713/wrapping-rate-limiting-api-call. Maybe you could help. Huge thanks! — SuperJMN, Aug 06 '16 at 14:03
I would prefer using a `Stopwatch` instead of `DateTime.UtcNow` for measuring intervals. It is more lightweight, and is guaranteed to always progress forward. The `DateTime.UtcNow` is subject to system-wise adjustments, and so it is possible to move backwards. — Theodor Zoulias, May 16 '20 at 23:22

How to build a rate-limiting API with Observables?

1 Answers1

Linked