streamでセマフォ？

semaphore-stream.js

var util      = require('util')
var stream    = require('stream')
var semaphore = require('semaphore')

function SemaphoreStream (count, option) {
    stream.Duplex.call(this, (option || {}))

    this.semaphore = semaphore(count)
    this.finished  = false

    var me = this

    this.once('finish', function onFinish () {
        me.finished = true
    })

    this.reverse = new Reverse

    this.reverse.on('error', function (err) {
        me.emit('error', err)
    })

    this.reverse.on('data', function reverserOnData (data) {
        me.semaphore.leave()
    })
}

util.inherits( SemaphoreStream, stream.Duplex )

SemaphoreStream.prototype._write = function (chnk, enc, cb) {
    var me = this
    this.semaphore.take(function () {
        me.push(chnk)
    })
    cb()
}

SemaphoreStream.prototype._read = function () {}

module.exports = SemaphoreStream


function Reverse () {
    stream.Transform.call(this)
}
util.inherits( Reverse, stream.Transform )
Reverse.prototype._transform = function (chnk, enc, cb) {
    this.push(chnk)
    cb()
}

timer.js

var path   = require('path')
var stream = require('stream')
var util   = require('util')
var Semaph = require(path.join( __dirname, './semaphore-stream'))

var queue = [
    'http://foo.org/bar'
  , 'http://bar.org/foo'
  , 'http://huh.org/huu'
  , 'http://foo.org/bar/bar'
  , 'http://bar.org/foo/foo'
  , 'http://huh.org/huu/huu'
]

var rs = new stream.Readable
rs._read = function () {
    var resource = queue.shift() || null
    this.push(resource)
}

var timer = new stream.Transform
timer.timeout = 1000
timer._transform = function (chnk, enc, cb) {
    var me = this
    setTimeout(function () {
        me.push(chnk)
    }, this.timeout)
    cb()
}

var sm = new Semaph(2)

rs.pipe(sm).pipe(timer, {end: false}).pipe(sm.reverse)
timer.pipe(process.stdout, {end: false})

end イベントがうまく起こせない...