throttle a stream with backpressure