add EventStream::queue method

[michaelficarra]

I'd like to propose a queue method for EventStreams. This method is used to consume only a given number of events from a stream simultaneously. When the consuming operation finishes (its stream is ended), it frees space for more consumption. Here's an implementation I'm currently using:

Bacon.EventStream::queue = (size, fn) ->
  buffer = []
  outStream = new Bacon.Bus
  draining = 0
  drain = ->
    ++draining
    stream = new Bacon.Bus
    outStream.plug stream
    stream.onEnd ->
      --draining
      if buffer.length > 0
        drain()
    fn buffer.shift(), stream
  @onError (e) ->
    outStream.error e
  @onValue (v) ->
    buffer.push v
    if draining < size
      drain()
    return
  outStream

Like flatMap, the given function returns a stream, the events of which get pushed to the stream produced by queue. Here's an example of my use of queue:

# get the email of the author responsible for changing a given host to/from a given hostclass
negligentAuthors = updatedHosts.queue 4, ([host, [runningHostclass, configuredHostclass]], stream) ->
  stdout = ''
  gitProcess = # long git invocation using child_process.spawn, omitted
  gitProcess.stdout.on 'data', (data) -> stdout += data
  gitProcess.stderr.pipe process.stderr
  gitProcess.on 'close', (exitCode) ->
    if exitCode is 0
      if stdout
        stream.push [stdout[...-1], host, runningHostclass, configuredHostclass]
    else
      stream.error exitCode
    stream.end()
  return

Without queue, processing this stream would cause a very large number of git processes to be running simultaneously, as it was when I was simply using flatMap. With queue, only at most 4 git processes are ever running simultaneously. Would you consider this function for inclusion?

[raimohanska]

I like the idea, and I feel that I too have some time ago felt the need for a combinator like this.

Some comments on the implementation though:

1. I'd rather have the fn return a stream instead of being passed a stream. This would eliminate the need for a Bus and be more consistent with, for example, flatMap. If fact, this can be seen as a flatMap variant with a limit for maximum concurrent spawned streams. In your example, you'd return either Bacon.never(), Bacon.once([ stdout blah blah ]) or Bacon.once(new Bacon.Error(exitCode)). It would be still nicer if it supported auto-wrapping like flatMap does, so that you could return [sdout..] and new Bacon.Error(exitCode) directly.
2. the outStream will not behave nicely, because it doesn't subscribe lazily. This requires a partial re-impl. Instead of creating a Bus, you should return a new EventStream(..) or use Bacon.bind(..) and so that laziness will be preserved

And, as for many other cases, I'm not sure where to put this. This should be resolved as part of #311.

EDIT: my detailed suggestions in (1) are actually incorrect. Instead you have to return a stream that outputs stuff on the gitProcess "close" event

[michaelficarra]

I'm afraid I'm just not experienced enough with Bacon to make those changes. They sound like good ideas, but I am unable to implement them. Bacon.EventStream is practically undocumented, and even looking at others' usage of it is no help.

[raimohanska]

I suggest using Bacon.fromBinder.

Did you already read my Wrapping Things in Bacon post? That might help.

[raimohanska]

I added a flatMapWithConcurrencyLimit method in the feature/flatMapWithConcurrencyLimit branch. Please try!

You should now be able to do like

# get the email of the author responsible for changing a given host to/from a given hostclass
negligentAuthors = updatedHosts.flatMapWithConcurrencyLimit 4, ([host, [runningHostclass, configuredHostclass]]) ->
  stdout = ''
  gitProcess = # long git invocation using child_process.spawn, omitted
  gitProcess.stdout.on 'data', (data) -> stdout += data
  gitProcess.stderr.pipe process.stderr
  Bacon.fromEventTarget(gitProcess, 'close').flatMap (exitCode) ->
    if exitCode is 0
      if stdout
        [stdout[...-1], host, runningHostclass, configuredHostclass]
      else
        Bacon.never()
    else
      new Bacon.Error(exitCode)

[raimohanska]

I think the new method is a great enabler. For instance limitThroughPut in #82 can now be implemented trivially on top of flatMapWithConcurrencyLimit.

[michaelficarra]

This is excellent! Exactly what I needed. Feel free to close this once you merge to master.