semaphoreで複雑な遅延ストリームを実装

guard-timer.js

'use strict'
var semaphore  = require('./semaphore')
module.exports = function guard (n) {
    var sem = semaphore(n)

    return function timer (f, timeout) {
        var id
        sem.wait(function () {
            id = setTimeout(cancel, timeout)
            f()
        })

        return cancel

        function cancel () {
            if (id) {
                clearTimeout(id)
                id = undefined
                sem.signal()
            }
        }
    }
}

semaphore.js

'use strict'
module.exports = function semaphore (n) {
    var capacity = n || 1
    var current  = 0
    var queue    = []

    return {
        wait:   wait
      , signal: signal
    }

    function wait (f) {
        ;(current < capacity) ? f() : queue.push(f)
        return (current += 1)
    }

    function signal () {
        if (current === 0) return
        current -= 1
        ;(typeof queue[0] === 'function') && queue.shift()()
    }
}

test.js

'use strict'
var es     = require('event-stream')
var timer  = require('./lib/timer-stream')

es.readArray(('AbcDefGhiJklMnoPqrStuVwxYz').split(''))
    .pipe(timer(3, 3000))
    .pipe(timer(1, 500))
    .pipe(es.map(function (cha, cb) { cb(null, cha + '\n') }))
    .pipe(process.stdout)

timer-stream.js

'use strict'
var es    = require('event-stream')
var guard = require('./guard-timer')

module.exports = function timerStream (n, timeout) {
    var g = guard(n)
    return es.map(function (value, cb) {
        g(function () { cb(null, value) }, timeout)
    })
}