Throttle / rateLimit

[richardscarrott]

Hi @reconbot

I wondered if you had considered adding a throttle utility which would yield values at a given rate. e.g.

function* numbers(max: number) {
    let i = 1;
    while (i <= max) {
      yield i++;
    }
  }

  await pipeline(
    () => numbers(10),
    throttle(
       2, // number of values to yield per window
       1000 // window of time to limit the values to (in ms)
    ),
    tap(async (num) => {
      console.log('Processing', num);
    }),
    consume
  );

// 0s Processing 1
// 0s Processing 2
// 1s Processing 3
// 1s Processing 4
// 2s Processing 5
// 2s Processing 6
// 3s Processing 7
// 3s Processing 8
// 4s Processing 9
// 4s Processing 10

My current use case is to process a stream of data which comes in faster than the third party API I'm sending it to can handle, so something like this would be useful, unless you can think of a better solution?

[richardscarrott]

For reference Highland (yeah, I'm basically copy pasting their stuff over at this rate 😅) has a ratelimit function which behaves the same way.

NOTE: Highland also has a throttle function, but that actually discards data which I can't imagine being too desirable for a pull based model?

Also, there's p-throttle which offers the same idea for promises.

I think the implementation would be something like this:

async function* _throttle<T>(
  num: number,
  ms: number,
  iterable: AnyIterable<T>
) {
  let sent = 0;
  let time: number | undefined;
  for await (const val of iterable) {
    if (sent < num) {
      if (typeof time === "undefined") {
        time = Date.now();
      }
      sent++;
      yield val;
      continue;
    }
    // Only wait if the window hasn't passed while yielding
    // the previous values.
    // NOTE: Highland doesn't make this optimisation, but I think it prob should have as if this
    // just waited for `ms` it'll likely over throttle if the `yield val` was slow.
    const elapsedMs = Date.now() - time!;
    const waitFor = ms - elapsedMs;
    time = undefined;
    waitFor > 0 && (await sleep(waitFor));
    if (typeof time === "undefined") {
      time = Date.now();
    }
    sent = 1;
    yield val;
  }
}

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

[reconbot]

Released in [email protected] https://github.com/reconbot/streaming-iterables/releases/tag/v6.2.0 🎉

[reconbot]

Thanks for getting it over the finish line =)