Turning hot observables to cold

Reactive-Extensions · Dec 28, 2013 · dd33b76 · dd33b76
1 parent e0f4709
commit dd33b76
Show file tree

Hide file tree

Showing 24 changed files with 330 additions and 345 deletions.
diff --git a/rx.aggregates.min.js b/rx.aggregates.min.js
diff --git a/rx.async.compat.js b/rx.async.compat.js
@@ -109,35 +109,34 @@
     Observable.fromCallback = function (func, scheduler, context, selector) {
         scheduler || (scheduler = timeoutScheduler);
         return function () {
-            var args = slice.call(arguments, 0), 
-                subject = new AsyncSubject();
+            var args = slice.call(arguments, 0);
 
-            scheduler.schedule(function () {
-                function handler(e) {
-                    var results = e;
-
-                    if (selector) {
-                        try {
-                            results = selector(arguments);
-                        } catch (err) {
-                            subject.onError(err);
-                            return;
+            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];
+                            }
                         }
-                    } else {
-                        if (results.length === 1) {
-                            results = results[0];
-                        }
-                    }
 
-                    subject.onNext(results);
-                    subject.onCompleted();
-                }
+                        observer.onNext(results);
+                        observer.onCompleted();
+                    }
 
-                args.push(handler);
-                func.apply(context, args);
+                    args.push(handler);
+                    func.apply(context, args);
+                });
             });
-
-            return subject.asObservable();
         };
     };
 
@@ -152,40 +151,40 @@
     Observable.fromNodeCallback = function (func, scheduler, context, selector) {
         scheduler || (scheduler = timeoutScheduler);
         return function () {
-            var args = slice.call(arguments, 0), 
-                subject = new AsyncSubject();
-
-            scheduler.schedule(function () {
-                function handler(err) {
-                    if (err) {
-                        subject.onError(err);
-                        return;
-                    }
+            var args = slice.call(arguments, 0);
 
-                    var results = slice.call(arguments, 1);
+            return new AnonymousObservable(function (observer) {
+                return scheduler.schedule(function () {
 
-                    if (selector) {
-                        try {
-                            results = selector(results);
-                        } catch (e) {
-                            subject.onError(e);
+                    function handler(err) {
+                        if (err) {
+                            observer.onError(err);
                             return;
                         }
-                    } else {
-                        if (results.length === 1) {
-                            results = results[0];
+
+                        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];
+                            }
                         }
-                    }
 
-                    subject.onNext(results);
-                    subject.onCompleted();
-                }
+                        observer.onNext(results);
+                        observer.onCompleted();
+                    }
 
-                args.push(handler);
-                func.apply(context, args);
+                    args.push(handler);
+                    func.apply(context, args);
+                });
             });
-
-            return subject.asObservable();
         };
     };
 
@@ -358,18 +357,16 @@
      * @returns {Observable} An Observable sequence which wraps the existing promise success and failure.
      */
     Observable.fromPromise = function (promise) {
-        var subject = new AsyncSubject();
-
-        promise.then(
-            function (value) {
-                subject.onNext(value);
-                subject.onCompleted();
-            }, 
-            function (reason) {
-               subject.onError(reason);
-            });
-
-        return subject.asObservable();
+        return new AnonymousObservable(function (observer) {
+            promise.then(
+                function (value) {
+                    observer.onNext(value);
+                    observer.onCompleted();
+                }, 
+                function (reason) {
+                   observer.onError(reason);
+                });
+        });
     };
     return Rx;
 }));