RxJS 3 CurrentFrame Scheduler

Generator.js

var noop = require("rx/support/noop");
var try_catch = require("rx/support/try-catch");
var flatten_args = require("rx/support/arguments-flatten");
var errorObj = require("rx/support/error-object");

function Generator(destinationOrNext, _throw, _return) {
    if(destinationOrNext && typeof destinationOrNext === "object") {
        this.result = destinationOrNext.result || { done: false };
        this.destination = destinationOrNext;
    } else {
        this.result = { done: false };
        this._next = destinationOrNext || noop;
        this._throw = _throw || noop;
        this._return = _return || noop;
    }
}

Generator["next"] = function generator_next(_next) {
    return new Generator(_next, null, null);
};

Generator["throw"] = function generator_throw(_throw) {
    return new Generator(null, _throw, null);
};

Generator["return"] = function generator_return(_return) {
    return new Generator(null, null, _return);
};

Generator.empty = function () { return new Generator(); };

Generator.prototype.length = 0;

Generator.prototype["next"] = function(value) {
    var result = this.result;
    if(Boolean(result.done)) {
        return result;
    }
    var result2 = try_catch(this._next).call(this, value) || result;
    if(result2 === errorObj) {
        throw errorObj.e;
    } else if(result !== result2 && typeof result2 === "object") {
        result.done = result2.done;
        result.value = result2.value;
    }
    if(result.done) {
        this.dispose();
    }
    return result;
};

Generator.prototype["throw"] = function(error) {
    var result = this.result;
    if(Boolean(result.done)) {
        return result;
    }
    var result2 = try_catch(this._throw).call(this, error) || result;
    if(result2 === errorObj) {
        throw errorObj.e;
    } else if(result !== result2 && typeof result2 === "object") {
        result.done = result2.done;
        result.value = result2.value;
    } else {
        result.done = true;
    }
    if(result.done) {
        this.dispose();
    }
    return result;
};

Generator.prototype["return"] = function(value) {
    var result = this.result;
    if(Boolean(result.done)) {
        return result;
    }
    var result2 = try_catch(this._return).call(this, value) || result;
    if(result2 === errorObj) {
        throw errorObj.e;
    } else if(result !== result2 && typeof result2 === "object") {
        result.done = result2.done;
        result.value = result2.value;
    } else {
        result.done = true;
    }
    if(result.done) {
        this.dispose();
    }
    return result;
};

Generator.prototype._next = function(value) {
    return this.destination["next"](value);
};

Generator.prototype._throw = function(error) {
    return this.destination["throw"](error);
};

Generator.prototype._return = function(value) {
    return this.destination["return"](value);
};

Generator.prototype.add = function add() {
    var result = this.result;
    var done = result.done;
    var value = result.value;
    var generators = this._generators || (this._generators = []);
    var additions = flatten_args(arguments);
    var index = -1;
    var count = additions.length;
    while(++index < count) {
        var generator = additions[index];
        if(!Boolean(generator)) {
            continue;
        }
        switch(typeof generator) {
            case "function":
                generator = Generator["return"](generator);
            case "object":
                if(generator && generator.result && !generator.result.done) {
                    if(done) {
                        generator["return"](value);
                    } else {
                        generators.push(generator);
                    }
                }
                break;
        }
    }
    this.length = generators.length;
    return this;
};

Generator.prototype.remove = function remove() {
    var done = this.result.done;
    var generators = this._generators;
    if(!done && generators && generators.length) {
        var subtractions = flatten_args(arguments);
        if(subtractions.length) {
            var index = -1;
            var count = subtractions.length;
            while(++index < count) {
                var generator = subtractions[index];
                var generatorIndex = generators.indexOf(generator);
                if(~generatorIndex) {
                    generators.splice(generatorIndex, 1);
                    if(generators.length === 0) {
                        break;
                    }
                }
            }
            this.length = generators.length;
        }
    }
    return this;
};

Generator.prototype.dispose = function dispose() {
    var generators = this._generators;
    if(generators && generators.length) {
        this._generators = undefined;
        var value = this.result.value;
        var index = -1;
        var count = generators.length;
        while(++index < count) {
            generators[index]["return"](value);
        }
        generators.length = 0;
    }
    return this;
};

module.exports = Generator;

Scheduler.js

var Generator = require("rx/Generator");
var Immediate = require("rx/support/Immediate");

var ScheduledAction = function () {

    function ScheduledAction(work, state, id, clear) {
        this.work = work;
        this.state = state;
        this.id = id;
        this.clear = clear;
        this.result = { done: false };
    }

    ScheduledAction.prototype = Object.create(Generator.prototype);

    ScheduledAction.prototype.result = { done: false };

    ScheduledAction.prototype["next"] = function scheduledActionNext(scheduler) {
        this.work(scheduler, this.state);
    };

    ScheduledAction.prototype["return"] = function scheduledActionReturn() {

        var id = this.id;
        var clear = this.clear;
        var tail = this.tail;
        var head = this.head;

        this.id = this.clear = this.work = this.state = void 0;

        tail && (tail.head = head);
        head && (head.tail = tail);

        if (typeof id !== "undefined") {
            clear(id);
        }

        return head;
    }

    return ScheduledAction;
}();

var Scheduler = function () {

    function Scheduler(async) {
        this.async = async || false;
        this.active = false;
        this.scheduled = false;
    }

    Scheduler.async = false;
    Scheduler.active = false;
    Scheduler.scheduled = false;

    Scheduler.schedule = Scheduler.prototype.schedule = function schedule(delay, state, work) {

        var argsLen = arguments.length;

        if (argsLen == 2) {
            work = state;
            state = delay;
            delay = 0;
        } else if (argsLen == 1) {
            work = delay;
            state = void 0,
            delay = 0;
        }

        if (delay <= 0) {
            if (Boolean(this.async)) {
                return scheduleNext(this, state, work);
            } else {
                return scheduleNow(this, state, work);
            }
        } else {
            return scheduleLater(this, delay, state, work);
        }
    };

    function scheduleNow(scheduler, state, work) {
        var curr = scheduler.head;
        var action = new ScheduledAction(work, state);
        if (Boolean(curr)) {
            action.tail = curr;
            curr.head = action;
        }
        scheduler.head = action;
        flush(scheduler, action);
        return action;
    }

    function scheduleNext(scheduler, state, work) {

        var action;

        if (!Boolean(scheduler.scheduled)) {
            scheduler.active = true;
            scheduler.scheduled = true;
            action = scheduleNow(scheduler, state, work);
            action.clear = Immediate.clearImmediate;
            action.id = Immediate.setImmediate(function () {
                action.id = void 0;
                scheduler.active = false;
                scheduler.scheduled = false;
                flush(scheduler, action);
            });
        } else {
            action = scheduleNow(scheduler, state, work);
        }

        return action;
    }

    function scheduleLater(scheduler, delay, state, work) {
        var action;
        var id = setTimeout(function () {
            id = void 0;
            action = scheduleNow(scheduler, state, work);
        }, delay);
        return new Generator(null, null, function () {
            if (id != null) {
                clearTimeout(id);
                id = void 0;
            }
            if (action != null) {
                if(scheduler.head === action) {
                    scheduler.head = void 0;
                }
                action.return();
            }
        });
    }

    function flush(scheduler, tail) {
        if (!Boolean(scheduler.active)) {
            scheduler.active = true;
            do {
                tail.next(scheduler);
            } while(tail = tail.return());
            scheduler.head = void 0;
            scheduler.active = false;
        }
    }

    Scheduler.scheduler = new Scheduler(true);

    return Scheduler;
}();

module.exports = Scheduler;