feat(facade): added support for observables

modules/angular2/package.json

@@ -9,6 +9,7 @@
   "dependencies": {
     "traceur": "<%= packageJson.dependencies.traceur %>",
     "rtts_assert": "<%= packageJson.version %>",
+    "rx": "<%= packageJson.dependencies['rx'] %>",
     "zone.js": "<%= packageJson.dependencies['zone.js'] %>"
   },
   "devDependencies": <%= JSON.stringify(packageJson.devDependencies) %>

modules/angular2/src/facade/async.dart

@@ -1,7 +1,7 @@
 library angular.core.facade.async;
 
 import 'dart:async';
-export 'dart:async' show Future;
+export 'dart:async' show Future, Stream, StreamController, StreamSubscription;
 
 class PromiseWrapper {
   static Future resolve(obj) => new Future.value(obj);
@@ -32,6 +32,32 @@ class PromiseWrapper {
   }
 }
 
+class ObservableWrapper {
+  static StreamSubscription subscribe(Stream s, Function onNext, [onError, onComplete]) {
+    return s.listen(onNext, onError: onError, onDone: onComplete, cancelOnError: true);
+  }
+
+  static StreamController createController() {
+    return new StreamController.broadcast();
+  }
+
+  static Stream createObservable(StreamController controller) {
+    return controller.stream;
+  }
+
+  static void callNext(StreamController controller, value) {
+    controller.add(value);
+  }
+
+  static void callThrow(StreamController controller, error) {
+    controller.addError(error);
+  }
+
+  static void callReturn(StreamController controller) {
+    controller.close();
+  }
+}
+
 class _Completer {
   final Completer c;
 

modules/angular2/src/facade/async.es6

@@ -1,5 +1,6 @@
-import {int, global} from 'angular2/src/facade/lang';
+import {int, global, isPresent} from 'angular2/src/facade/lang';
 import {List} from 'angular2/src/facade/collection';
+import Rx from 'rx/dist/rx.all';
 
 export var Promise = global.Promise;
 
@@ -51,3 +52,47 @@ export class PromiseWrapper {
     return maybePromise instanceof Promise;
   }
 }
+
+
+/**
+ * Use Rx.Observable but provides an adapter to make it work as specified here:
+ * https://github.com/jhusain/observable-spec
+ *
+ * Once a reference implementation of the spec is available, switch to it.
+ */
+export var Observable = Rx.Observable;
+export var ObservableController = Rx.Subject;
+
+export class ObservableWrapper {
+  static createController():Rx.Subject {
+    return new Rx.Subject();
+  }
+
+  static createObservable(subject:Rx.Subject):Observable {
+    return subject;
+  }
+
+  static subscribe(observable:Observable, generatorOrOnNext, onThrow = null, onReturn = null) {
+    if (isPresent(generatorOrOnNext.next)) {
+      return observable.observeOn(Rx.Scheduler.timeout).subscribe(
+        (value) => generatorOrOnNext.next(value),
+        (error) => generatorOrOnNext.throw(error),
+        () => generatorOrOnNext.return()
+      );
+    } else {
+      return observable.observeOn(Rx.Scheduler.timeout).subscribe(generatorOrOnNext, onThrow, onReturn);
+    }
+  }
+
+  static callNext(subject:Rx.Subject, value:any) {
+    subject.onNext(value);
+  }
+
+  static callThrow(subject:Rx.Subject, error:any) {
+    subject.onError(error);
+  }
+
+  static callReturn(subject:Rx.Subject) {
+    subject.onCompleted();
+  }
+}

modules/angular2/test/facade/async_spec.js

@@ -0,0 +1,99 @@
+import {describe, it, expect, beforeEach, ddescribe, iit, xit, el,
+  SpyObject, AsyncTestCompleter, inject, IS_DARTIUM} from 'angular2/test_lib';
+
+import {ObservableWrapper, Observable, ObservableController, PromiseWrapper} from 'angular2/src/facade/async';
+
+export function main() {
+  describe('Observable', () => {
+    var obs:Observable;
+    var controller:ObservableController;
+
+    beforeEach(() => {
+      controller = ObservableWrapper.createController();
+      obs = ObservableWrapper.createObservable(controller);
+    });
+
+    it("should call the next callback",  inject([AsyncTestCompleter], (async) => {
+      ObservableWrapper.subscribe(obs, (value) => {
+        expect(value).toEqual(99);
+        async.done();
+      });
+
+      ObservableWrapper.callNext(controller, 99);
+    }));
+
+    it("should call the throw callback", inject([AsyncTestCompleter], (async) => {
+      ObservableWrapper.subscribe(obs, (_) => {}, (error) => {
+        expect(error).toEqual("Boom");
+        async.done();
+      });
+      ObservableWrapper.callThrow(controller, "Boom");
+    }));
+
+    it("should call the return callback", inject([AsyncTestCompleter], (async) => {
+      ObservableWrapper.subscribe(obs, (_) => {}, (_) => {}, () => {
+        async.done();
+      });
+
+      ObservableWrapper.callReturn(controller);
+    }));
+
+    it("should subscribe to the wrapper asynchronously", () => {
+      var called = false;
+      ObservableWrapper.subscribe(obs, (value) => {
+        called = true;
+      });
+
+      ObservableWrapper.callNext(controller, 99);
+      expect(called).toBe(false);
+    });
+
+    if (!IS_DARTIUM) {
+      // See here: https://github.com/jhusain/observable-spec
+      describe("Generator", () => {
+        var generator;
+
+        beforeEach(() => {
+          generator = new SpyObject();
+          generator.spy("next");
+          generator.spy("throw");
+          generator.spy("return");
+        });
+
+        it("should call next on the given generator",  inject([AsyncTestCompleter], (async) => {
+          generator.spy("next").andCallFake((value) => {
+            expect(value).toEqual(99);
+            async.done();
+          });
+
+          ObservableWrapper.subscribe(obs, generator);
+          ObservableWrapper.callNext(controller, 99);
+        }));
+
+        it("should call throw on the given generator", inject([AsyncTestCompleter], (async) => {
+          generator.spy("throw").andCallFake((error) => {
+            expect(error).toEqual("Boom");
+            async.done();
+          });
+          ObservableWrapper.subscribe(obs, generator);
+          ObservableWrapper.callThrow(controller, "Boom");
+        }));
+
+        it("should call return on the given generator", inject([AsyncTestCompleter], (async) => {
+          generator.spy("return").andCallFake(() => {
+            async.done();
+          });
+          ObservableWrapper.subscribe(obs, generator);
+          ObservableWrapper.callReturn(controller);
+        }));
+      });
+    }
+
+    //TODO: vsavkin: add tests cases
+    //should call dispose on the subscription if generator returns {done:true}
+    //should call dispose on the subscription on throw
+    //should call dispose on the subscription on return
+ });
+}
+
+//make sure rx observables are async