Add Stream::throttle

[yoshuawuyts]

We should add a Stream::throttle method as "unstable", similar to RxJs's throttle function.

The goal of this function is to limit the amount of items yielded per timeslice in a stream. This is usually considered a somewhat brute tool, but it does have its use and we should give people access to this.

Examples

// emit value every 1 second
let s = stream::interval(Duration::from_secs(1)).enumerate();

// throttle for 2 seconds, emit latest value
// in this case it means we skip two iterations, and let through the third
let s = s.throttle(Duration::from_secs(2));

s.for_each(|(_, n)| dbg!(n)).await;
// => 0 .. 3 .. 6 .. 9

References

• https://www.learnrxjs.io/operators/filtering/throttle.html
• https://rxjs.dev/api/operators/throttle

[Wassasin]

Made something I'll finish later that needs some testing. We might also consider implementing debounce.

[binarybana]

In my mind the operation described is more of a "subsampling" operation, whereas throttle applies backpressure but would otherwise be non lossy. But I'm just bike shedding

[yoshuawuyts]

@Wassasin hehe, yeah I wanted to add debounce to the next batch of issues. But this is definitely something we would want, so feel free to go for it!

@binarybana hmm, you might be right. I do think that dropping events is the right approach for any throttle function, but it may make sense for the event source to determine how it sheds load (e.g. LIFO vs FIFO queueing, bounded vs unbounded, max queue size, etc.) — in that case your assessment is accurate and we should make throttle apply backpressure only.

[mikecaines]

I implemented the stream-throttle crate, which uses a backpressure approach.

It uses an intermediate ThrottlePool type, so that you can throttle multiple streams together. Similar functionality (but not identical) can be achieved with Stream::select().

Perhaps some part of it could form the base for a solution. Just a thought!