bacon.jsで関数型プログラミングをやってみる(2) - Qiita http://qiita.com/m4fg/items/9b0d5261909224fd45d0 を node.jsのストリームでやってみる

guard-timer-stream.js

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

module.exports = function (/*capacity, msec*/) {
    var args     = [].slice.apply(arguments)
    var msec     = args.pop()
    var capacity = args.shift()
    var timer    = guard(capacity)

    return es.map(function (data, cb) {
        timer(function () {cb(null, data)} , msec)
    })
}

guard-timer.js

'use strict'
var semaphore = require('./semaphore')
// var timer = guard(3)
// var cancel = timer(function () {
//     consle.log((new Date()).toUTCString())
// }, 1000)
module.exports = function guard (n) {
    var sem = semaphore(n)

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

        return cancel

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

scrape.js

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

var list = [
  'http://qiita.com/advent-calendar/2011',
  'http://qiita.com/advent-calendar/2012',
  'http://qiita.com/advent-calendar/2013',
  'http://qiita.com/advent-calendar/2014'
]

function fetchFromUrl () {
    return es.map(function (url, cb) {
        client.fetch(url, {}, function (err, $, res) {
            cb(null, $)
        })
    })
}

function getTitle () {
    return es.map(function ($, cb) {
        cb(null, $('title').text())
    })
}

es.readArray(list)
  .pipe(timers(1000))   // 1秒ごとに url を流す
  .pipe(fetchFromUrl()) // 取得結果（をハンドルする$）を流す
  .pipe(getTitle())     // titleを取得して流す
  .pipe(es.stringify()) // 改行を端折るために手抜き
  .pipe(process.stdout) // 標準出力

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)
        current += 1
    }

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