bacon has no real throttling/rate-limiting method

[michaelficarra]

Bacon streams have a load of buffering methods, but are strangely missing the one I actually want to use. For comparison purposes, I'll include the diagrams from the documentation here:

var delayed = source.delay(2);
// source:    asdf----asdf----
// delayed:   --asdf----asdf--

var throttled = source.throttle(2);
// source:    asdf----asdf----
// throttled: --s--f----s--f--

var debounced = source.debounce(2);
// source:    asdf----asdf----
// debounced: -----f-------f--

var debounced = source.debounceImmediate(2);
// source:    asdf----asdf----
// debounced: a-d-----a-d-----

What I'm looking for is a function that preserves all events, allowing them to enter the produced stream with a minimum delay between each event.

var limited = source.rateLimit(2);
// source:  asdf----asdf----
// limited: a-s-d-f-a-s-d-f-

The closest I could find was a mention of limitThroughput in #82, which seems not to have been added. This is the implementation I'm using right now:

setTimeout_ = (ms, fn) -> setTimeout fn, ms

bacon.EventStream::rateLimit = (ms) ->
  buffer = []
  outStream = new Bacon.Bus
  draining = no
  drain = ->
    outStream.push buffer.shift()
    setTimeout_ ms, ->
      if buffer.length > 0
        drain()
      else
        draining = no
    return
  @onError (e) ->
    outStream.error e
  @onValue (e) ->
    buffer.push e
    unless draining
      draining = yes
      drain()
    return
  outStream

I know it's probably not very idiomatic, since I am new to bacon, but it is working for me.

[raimohanska]

As I mentioned in #318, rateLimit / limitThroughPut is trivial to implement using flatMapWithConcurrencyLimit. The implementation for both is now available for testing in the feature/flatMapWithConcurrencyLimit branch.

[michaelficarra]

Unfortunately, this implementation of rateLimit forces the operations to be sequential. stream.rateLimit(0).flatMap ... should be equivalent to stream.flatMap ..., but it is not behaving that way for me.

edit: Though I've just realised that my implementation above doesn't work this way either.

[raimohanska]

Streams are sequential by nature. The rateLimit combinator just limits the throughput of the stream.

The flatMapWithConcurrencyLimit is the one that allows you to run a limited number of "substreams" in parallel, collecting the results of each into a new (once again sequential) stream of events.

[raimohanska]

What are you actually looking for? :)

[michaelficarra]

Sorry for describing this feature request so poorly. I'll try to do a better job here. I'm reading a directory and dumping each file name into a stream. For each file name, I am kicking off an HTTP request that, when it finishes, will populate another stream with one or more events. I'm trying to be a nice citizen on our network and limit the number of requests made per second. So even though the file name events happen very quickly and the HTTP requests may take a while to respond, I want to start them at a rate no faster than (in my case) one every 100ms.

[raimohanska]

So you do

makeRequest = (filename) -> { url: "/stuff/" + filename }
results = filenames.rateLimit(100).map(makeRequest).ajax()

And you'll be making new requests at most once in 100ms. If you're using Bacon.JQuery, you'll get parallel response streams merged into one. Are you?

[phadej]

I would use something like

throttler = Bacon.interval(100, "really anything");
results = filenames.zip(throttler, (a, b) -> a).map(makeRequest).ajax()

but this of course will not behave well enough, if there are long quiet periods, throttler will produce new values and leak space... and allow burst after quiet periods...

However, maybe some kind of biased zip (which buffers only other param) would be more generic?

[raimohanska]

I don't see why you wouldn't use the new rateLimit method which seems like the perfect fit :)