rx.async.js

// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.

;(function (factory) {
    var objectTypes = {
        'boolean': false,
        'function': true,
        'object': true,
        'number': false,
        'string': false,
        'undefined': false
    };

    var root = (objectTypes[typeof window] && window) || this,
        freeExports = objectTypes[typeof exports] && exports && !exports.nodeType && exports,
        freeModule = objectTypes[typeof module] && module && !module.nodeType && module,
        moduleExports = freeModule && freeModule.exports === freeExports && freeExports,
        freeGlobal = objectTypes[typeof global] && global;
    
    if (freeGlobal && (freeGlobal.global === freeGlobal || freeGlobal.window === freeGlobal)) {
        root = freeGlobal;
    }

    // Because of build optimizers
    if (typeof define === 'function' && define.amd) {
        define(['rx', 'exports'], function (Rx, exports) {
            root.Rx = factory(root, exports, Rx);
            return root.Rx;
        });
    } else if (typeof module === 'object' && module && module.exports === freeExports) {
        module.exports = factory(root, module.exports, require('./rx'));
    } else {
        root.Rx = factory(root, {}, root.Rx);
    }
}.call(this, function (root, exp, Rx, undefined) {
    
    // Aliases
    var Observable = Rx.Observable,
    	observableProto = Observable.prototype,
        AnonymousObservable = Rx.Internals.AnonymousObservable,
        AsyncSubject = Rx.AsyncSubject,
        disposableCreate = Rx.Disposable.create,
        CompositeDisposable= Rx.CompositeDisposable,
        immediateScheduler = Rx.Scheduler.immediate,
        slice = Array.prototype.slice;

    /**
     * Invokes the specified function asynchronously on the specified scheduler, surfacing the result through an observable sequence.
     * 
     * @example
     * var res = Rx.Observable.start(function () { console.log('hello'); });
     * var res = Rx.Observable.start(function () { console.log('hello'); }, Rx.Scheduler.timeout);
     * var res = Rx.Observable.start(function () { this.log('hello'); }, Rx.Scheduler.timeout, console);
     * 
     * @param {Function} func Function to run asynchronously.
     * @param {Scheduler} [scheduler]  Scheduler to run the function on. If not specified, defaults to Scheduler.timeout.
     * @param [context]  The context for the func parameter to be executed.  If not specified, defaults to undefined.
     * @returns {Observable} An observable sequence exposing the function's result value, or an exception.
     * 
     * Remarks
     * * The function is called immediately, not during the subscription of the resulting sequence.
     * * Multiple subscriptions to the resulting sequence can observe the function's result.  
     */
    Observable.start = function (func, scheduler, context) {
        return observableToAsync(func, scheduler, context)();
    };

    /**
     * Converts the function into an asynchronous function. Each invocation of the resulting asynchronous function causes an invocation of the original synchronous function on the specified scheduler.
     * 
     * @example
     * var res = Rx.Observable.toAsync(function (x, y) { return x + y; })(4, 3);
     * var res = Rx.Observable.toAsync(function (x, y) { return x + y; }, Rx.Scheduler.timeout)(4, 3);
     * var res = Rx.Observable.toAsync(function (x) { this.log(x); }, Rx.Scheduler.timeout, console)('hello');
     * 
     * @param {Function} function Function to convert to an asynchronous function.
     * @param {Scheduler} [scheduler] Scheduler to run the function on. If not specified, defaults to Scheduler.timeout.
     * @param {Mixed} [context] The context for the func parameter to be executed.  If not specified, defaults to undefined.
     * @returns {Function} Asynchronous function.
     */
    var observableToAsync = Observable.toAsync = function (func, scheduler, context) {
        scheduler || (scheduler = timeoutScheduler);
        return function () {
            var args = arguments, 
                subject = new AsyncSubject();

            scheduler.schedule(function () {
                var result;
                try {
                    result = func.apply(context, args);
                } catch (e) {
                    subject.onError(e);
                    return;
                }
                subject.onNext(result);
                subject.onCompleted();
            });
            return subject.asObservable();
        };
    };

    /**
     * Converts a callback function to an observable sequence. 
     * 
     * @param {Function} function Function with a callback as the last parameter to convert to an Observable sequence.
     * @param {Scheduler} [scheduler] Scheduler to run the function on. If not specified, defaults to Scheduler.timeout.
     * @param {Mixed} [context] The context for the func parameter to be executed.  If not specified, defaults to undefined.
     * @param {Function} [selector] A selector which takes the arguments from the callback to produce a single item to yield on next.
     * @returns {Function} A function, when executed with the required parameters minus the callback, produces an Observable sequence with a single value of the arguments to the callback as an array.
     */
    Observable.fromCallback = function (func, scheduler, context, selector) {
        scheduler || (scheduler = immediateScheduler);
        return function () {
            var args = slice.call(arguments, 0);

            return new AnonymousObservable(function (observer) {
                return scheduler.schedule(function () {
                    function handler(e) {
                        var results = e;
                        
                        if (selector) {
                            try {
                                results = selector(arguments);
                            } catch (err) {
                                observer.onError(err);
                                return;
                            }
                        } else {
                            if (results.length === 1) {
                                results = results[0];
                            }
                        }

                        observer.onNext(results);
                        observer.onCompleted();
                    }

                    args.push(handler);
                    func.apply(context, args);
                });
            });
        };
    };

    /**
     * Converts a Node.js callback style function to an observable sequence.  This must be in function (err, ...) format.
     * @param {Function} func The function to call
     * @param {Scheduler} [scheduler] Scheduler to run the function on. If not specified, defaults to Scheduler.timeout.
     * @param {Mixed} [context] The context for the func parameter to be executed.  If not specified, defaults to undefined.
     * @param {Function} [selector] A selector which takes the arguments from the callback minus the error to produce a single item to yield on next.     
     * @returns {Function} An async function which when applied, returns an observable sequence with the callback arguments as an array.
     */
    Observable.fromNodeCallback = function (func, scheduler, context, selector) {
        scheduler || (scheduler = immediateScheduler);
        return function () {
            var args = slice.call(arguments, 0);

            return new AnonymousObservable(function (observer) {
                return scheduler.schedule(function () {
                    
                    function handler(err) {
                        if (err) {
                            observer.onError(err);
                            return;
                        }

                        var results = slice.call(arguments, 1);
                        
                        if (selector) {
                            try {
                                results = selector(results);
                            } catch (e) {
                                observer.onError(e);
                                return;
                            }
                        } else {
                            if (results.length === 1) {
                                results = results[0];
                            }
                        }

                        observer.onNext(results);
                        observer.onCompleted();
                    }

                    args.push(handler);
                    func.apply(context, args);
                });
            });
        };
    };

    function createListener (element, name, handler) {
        // Node.js specific
        if (element.addListener) {
            element.addListener(name, handler);
            return disposableCreate(function () {
                element.removeListener(name, handler);
            });
        } else if (element.addEventListener) {
            element.addEventListener(name, handler, false);
            return disposableCreate(function () {
                element.removeEventListener(name, handler, false);
            });
        }
    }

    function createEventListener (el, eventName, handler) {
        var disposables = new CompositeDisposable();

        // Asume NodeList
        if (el && el.length) {
            for (var i = 0, len = el.length; i < len; i++) {
                disposables.add(createEventListener(el[i], eventName, handler));
            }
        } else if (el) {
            disposables.add(createListener(el, eventName, handler));
        }

        return disposables;
    }

    /**
     * Creates an observable sequence by adding an event listener to the matching DOMElement or each item in the NodeList.
     *
     * @example
     *   var source = Rx.Observable.fromEvent(element, 'mouseup');
     * 
     * @param {Object} element The DOMElement or NodeList to attach a listener.
     * @param {String} eventName The event name to attach the observable sequence.
     * @param {Function} [selector] A selector which takes the arguments from the event handler to produce a single item to yield on next.     
     * @returns {Observable} An observable sequence of events from the specified element and the specified event.
     */
    Observable.fromEvent = function (element, eventName, selector) {
        return new AnonymousObservable(function (observer) {
            return createEventListener(
                element, 
                eventName, 
                function handler (e) { 
                    var results = e;

                    if (selector) {
                        try {
                            results = selector(arguments);
                        } catch (err) {
                            observer.onError(err);
                            return
                        }
                    }

                    observer.onNext(results); 
                });
        }).publish().refCount();
    };
    /**
     * Creates an observable sequence from an event emitter via an addHandler/removeHandler pair.
     * @param {Function} addHandler The function to add a handler to the emitter.
     * @param {Function} [removeHandler] The optional function to remove a handler from an emitter.
     * @param {Function} [selector] A selector which takes the arguments from the event handler to produce a single item to yield on next.
     * @returns {Observable} An observable sequence which wraps an event from an event emitter
     */
    Observable.fromEventPattern = function (addHandler, removeHandler, selector) {
        return new AnonymousObservable(function (observer) {
            function innerHandler (e) {
                var result = e;
                if (selector) {
                    try {
                        result = selector(arguments);
                    } catch (err) {
                        observer.onError(err);
                        return;
                    }
                }
                observer.onNext(result);
            }

            var returnValue = addHandler(innerHandler);
            return disposableCreate(function () {
                if (removeHandler) {
                    removeHandler(innerHandler, returnValue);
                }
            });
        }).publish().refCount();
    };

    /**
     * Converts a Promise to an Observable sequence
     * @param {Promise} A Promises A+ implementation instance.
     * @returns {Observable} An Observable sequence which wraps the existing promise success and failure.
     */
    Observable.fromPromise = function (promise) {
        return new AnonymousObservable(function (observer) {
            promise.then(
                function (value) {
                    observer.onNext(value);
                    observer.onCompleted();
                }, 
                function (reason) {
                   observer.onError(reason);
                });
        });
    };
    /*
     * Converts an existing observable sequence to an ES6 Compatible Promise
     * @example
     * var promise = Rx.Observable.return(42).toPromise(RSVP.Promise);
     * @param {Function} The constructor of the promise
     * @returns {Promise} An ES6 compatible promise with the last value from the observable sequence.
     */
    observableProto.toPromise = function (promiseCtor) {
        var source = this;
        return new promiseCtor(function (resolve, reject) {
            // No cancellation can be done
            var value, hasValue = false;
            source.subscribe(function (v) {
                value = v;
                hasValue = true;
            }, function (err) {
                reject(err);
            }, function () {
                if (hasValue) {
                    resolve(value);
                }
            });
        });
    };
    return Rx;
}));