Reactに寄せた Flux/Store のプロトタイプ

observer.js

module.exports = Observer

function Observer () {
    this.subs = []
}

Observer.prototype.publish = function () {
    var args = [].slice.apply(arguments)
    for (var i = 0; i < this.subs.length; i++) {
        this.subs[i].apply(null, args)
    }
}

Observer.prototype.subscribe = function (sub) {
    return ('function' === typeof sub) ? !!(this.subs.push(sub)) : false
}

Observer.prototype.unsubscribe = function (sub) {
    this.subs = ('undefined' === typeof sub || null === sub)
        ? [] : this.subs.filter(function (_sub) {return _sub !== sub})
    return sub
}

Observer.prototype.once = function (sub) {
    var me = this
    function wrap () {
        sub.apply(null, arguments)
        me.unsubscribe(wrap)
    }
    this.subscribe(wrap)
}

semaphore.js

module.exports = function createSemaphore (_capacity) {
    var capacity = _capacity || 1
    var current  = 0
    var queue    = []

    var semaphore = {
        wait: wait
      , signal: signal
    }

    return semaphore

    function wait (f) {
        if (current < capacity) f()
        else queue.push(f)
        current += 1
    }

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

store.js

var xtend     = require('xtend')
var mutable   = require('xtend/mutable')
var clone     = require('clone')
var inherits  = require('inherits')
var isEmpty   = require('is-empty')
var deepEqual = require('deep-equal')
var Observer  = require('./lib/observer')
var semaphore = require('./lib/semaphore')

inherits(Store, Observer)
module.exports = Store

function Store (initState, children, _opt, _role) {
    if (!(this instanceof Store))
        return new Store(initState, children, _opt, _role)

    Observer.call(this)
    this.children  = children
    this.state     = initState
    this.opt       = xtend(_opt)
    this.deepEqual = deepEqual
    this.semaphore = semaphore(1)
    _role && mutable(this, _role)
}

Store.prototype.getState = function getState () {
    return clone(this.state)
}

Store.prototype.setState = function setState (_state) {
    var state = xtend(this.state, _state)
    if (this.deepEqual(state, this.state)) return false

    this.state = state
    return true
}

Store.prototype.post = function post (data, _doPublishFlg) {
    var me = this
    var doPublishFlg = !! _doPublishFlg
    this.semaphore.wait(function () {
        if (isEmpty(me.children)) _work(data, me.getState())

        else {
            var state
            var n   = 0
            var len = me.children.length;
            me.children.forEach(function (store) {
                store.once(function () {
                    state = xtend(state, store.getState())
                    if ((n += 1) === len) _work(state, me.getState())
                })
                store.post(data)
            })
        }
    })

    function _work (data, state) {
        me.work(data, state, function (err, _state) {
            var isUpdated
            if (err) me.error(err)
            else (isUpdated = me.setState(_state))

            if (doPublishFlg) (isUpdated && me.publish())
            else me.publish()

            me.semaphore.signal()
        })
    }
}

Store.prototype.work = function (data, state, done) {
    throw new Error('".work" has not been implemented.')
}

Store.prototype.error = function (err) {
    this.handleError(err)
}

Store.prototype.addHandleError = function (handleError) {
    this.handleError = handleError
}

example.

var xtend = require('xtend')
var sum = store({sum: 0}, null, null, {work: function (data, nowState, done) {
    done(null, {sum: nowState.sum + data})
}})
var count = store({count: 0}, null, null, {work: function (data, nowState, done) {
    setTimeout(function () {
        done(null, {count: nowState.count + 1})
    }, 0)
}})
var average = store({average: 0}, [sum, count], null, {work: function (sumAndCount, nowState, done) {
    done(null, xtend(sumAndCount, {average: sumAndCount.sum / sumAndCount.count}))
}})
var container = store({}, [average], null, {work: function (newState, nowState, done) { done(null, newState) })
// 
container.subscribe(function () {
    console.dir(container.getState())
})
container.addHandleError(function (err) {
    console.error(err)
})
//
container.post(3, true)
// {count: 1, sum: 3, average: 3}