markov_merge_q.js

Markov.prototype.merge = withReady(function merge(other, callback) {
    var self = this;
    if (!other.ready) {
        return other.init(function(err) {
            if (err) return callback(err);
            self.merge(other, callback);
        });
    }
    if (this.stateSize !== other.stateSize) {
        return callback(new Error('cannot merge markovs with differing state size'));
    }
    var Q = [], adding = false, finished = false;
    other.db.createReadStream().on('data', function(data) {
        if (data.key.indexOf('transitions/') === 0) {
            add(data.key, data.value);
        }
    }).on('error', finish).on('end', finish);
    function add(state, wTokens) {
        if (adding) {
            Q.push([state, wTokens]);
        } else {
            adding = true;
            self.addWeightedTransitions(state, wTokens, function(err) {
                adding = false;
                if (err) return finish(err);
                if (Q.length) add.apply(null, Q.shift());
            });
        }
    }
    function finish() {
        if (finished) return;
        finished = true;
        callback.apply(this, arguments);
    }
});