Using Bacon streams with resources

streams.coffee

thing1 =
  m_a: (list) ->
    console.log "Processing [#{list.join ',' }]"
    stream_a = new Bacon.Bus
    stream_b = thing2.m_b(stream_a)
    next = ->
      return unless list.length
      setTimeout (-> stream_a.push list.shift()), 300
    stream_b.onValue (result) ->
      console.log "Got result #{result}"
      next()
    next()

thing2 =
  queue: []
  locked: no
  addToQueue: (n, fn) ->
    fn2 = =>
      @processQueue()
      fn.apply null, arguments
    @queue.push [n, fn2]
    @processQueue()
  processQueue: ->
    return unless @queue.length and not @locked
    @locked = yes
    [n, fn] = @queue.shift()
    console.log("Processing input #{n}")
    setTimeout( =>
      @locked = no
      fn([n,n+1].join ' => ')
    , 300)
  m_b: (stream_a) ->
    stream_b = new Bacon.Bus
    stream_a.onValue (n) =>
      console.log "Stream A passed #{n}"
      @addToQueue n, (result) ->
        stream_b.push result
    stream_a.onEnd -> stream_b.push new Bacon.End
    return stream_b
  
thing1.m_a [2,3,5,8,13]