feat(from): let from handle any "observablesque"

- `Observable.from(promise)` should work
- `Observable.from(iterable)` should work
- `Observable.from(observable)` should work

This includes any object that implements `Symbol.observer`.

closes #156
closes #236

spec/observables/from-spec.js

@@ -1,6 +1,8 @@
-/* globals describe, it, expect */
+/* globals describe, it, expect, Symbol */
 var Rx = require('../../dist/cjs/Rx');
+var Promise = require('promise');
 var Observable = Rx.Observable;
+var $$iterator = require('../../dist/cjs/util/Symbol_iterator').default;
 
 describe('Observable.from', function () {
  it('should enumerate an Array', function (done) {
@@ -10,4 +12,55 @@ describe('Observable.from', function () {
  expect(x).toBe(expected[i++])
  }, null, done);
  }, 300);
+
+ it('should handle a promise', function (done) {
+ var promise = Promise.resolve('pinky swear');
+
+ Observable.from(promise).subscribe(function (x) {
+ expect(x).toBe('pinky swear');
+ }, null, done);
+ });
+
+ it('should handle an "observableque" object', function (done) {
+ var observablesque = {};
+ observablesque[Symbol.observer] = function (observer) {
+ observer.next('test');
+ observer.complete();
+ };
+
+ Observable.from(observablesque).subscribe(function (x) {
+ expect(x).toBe('test');
+ }, null, done);
+ });
+
+ it('should handle a string', function (done) {
+ var expected = ['a', 'b', 'c'];
+ Observable.from('abc').subscribe(function (x) {
+ expect(x).toBe(expected.shift());
+ }, null, done);
+ });
+
+ it('should handle any iterable thing', function (done) {
+ var iterable = {};
+ var iteratorResults = [
+ { value: 'one', done: false },
+ { value: 'two', done: false },
+ { done: true }
+ ];
+ var expected = ['one', 'two'];
+
+ expect(Symbol.iterator).toBe($$iterator);
+
+ iterable[Symbol.iterator] = function () {
+ return {
+ next: function() {
+ return iteratorResults.shift();
+ }
+ }
+ };
+
+ Observable.from(iterable).subscribe(function (x) {
+ expect(x).toBe(expected.shift());
+ }, null, done);
+ });
 });

src/Observable.ts

@@ -74,7 +74,7 @@ export default class Observable<T> {
  // TODO: convert this to an `abstract` class in TypeScript 1.6.
 
  static defer: <T>(observableFactory: () => Observable<T>) => Observable<T>;
- static from: <T>(iterable: any, project?: (x?: any, i?: number) => T, thisArg?: any, scheduler?: Scheduler) => Observable<T>;
+ static from: <T>(iterable: any, scheduler?: Scheduler) => Observable<T>;
  static fromArray: <T>(array: T[], scheduler?: Scheduler) => Observable<T>;
  static fromEvent: <T>(element: any, eventName: string, selector: (...args:Array<any>) => T) => Observable<T>;
  static fromEventPattern: <T>(addHandler: (handler:Function)=>void, removeHandler: (handler:Function) => void, selector?: (...args:Array<any>) => T) => Observable<T>;

src/Rx.ts

@@ -15,17 +15,17 @@ import EmptyObservable from './observables/EmptyObservable';
 import ErrorObservable from './observables/ErrorObservable';
 import InfiniteObservable from './observables/InfiniteObservable';
 import IntervalObservable from './observables/IntervalObservable';
-import IteratorObservable from './observables/IteratorObservable';
 import PromiseObservable from './observables/PromiseObservable';
 import RangeObservable from './observables/RangeObservable';
 import ScalarObservable from './observables/ScalarObservable';
 import TimerObservable from './observables/TimerObservable';
 import FromEventPatternObservable from './observables/FromEventPatternObservable';
 import FromEventObservable from './observables/FromEventObservable';
 import ForkJoinObservable from './observables/ForkJoinObservable';
+import FromObservable from './observables/FromObservable';
 
 Observable.defer = DeferObservable.create;
-Observable.from = IteratorObservable.create;
+Observable.from = FromObservable.create;
 Observable.fromArray = ArrayObservable.create;
 Observable.fromPromise = PromiseObservable.create;
 Observable.of = ArrayObservable.of;

src/observables/FromObservable.ts

@@ -0,0 +1,43 @@
+import PromiseObservable from './PromiseObservable';
+import IteratorObservable from'./IteratorObservable';
+import ArrayObservable from './ArrayObservable';
+
+import isArray from '../util/isArray';
+import isPromise from '../util/isPromise';
+import isObservable from '../util/isObservable';
+import Scheduler from '../Scheduler';
+import $$observer from '../util/Symbol_observer';
+import Observable from '../Observable';
+import Subscriber from '../Subscriber';
+import { ObserveOnSubscriber } from '../operators/observeOn';
+
+export default class FromObservable<T> extends Observable<T> {
+ constructor(private observablesque: any, private scheduler: Scheduler) {
+ super(null);
+ }
+
+ static create<T>(observablesque: any, scheduler: Scheduler = Scheduler.immediate): Observable<T> {
+ if (isArray(observablesque)) {
+ return new ArrayObservable(observablesque, scheduler);
+ } else if (isPromise(observablesque)) {
+ return new PromiseObservable(observablesque, scheduler);
+ } else if (isObservable(observablesque)) {
+ if(observablesque instanceof Observable) {
+ return observablesque;
+ }
+ return new FromObservable(observablesque, scheduler);
+ } else {
+ return new IteratorObservable(observablesque, null, null, scheduler);
+ }
+ }
+
+ _subscribe(subscriber: Subscriber<T>) {
+ const observablesque = this.observablesque;
+ const scheduler = this.scheduler;
+ if(scheduler === Scheduler.immediate) {
+ return this.observablesque[$$observer](subscriber);
+ } else {
+ return this.observablesque[$$observer](new ObserveOnSubscriber(subscriber, scheduler, 0));
+ }
+ }
+}

src/observables/IteratorObservable.ts

@@ -2,7 +2,7 @@ import Scheduler from '../Scheduler';
 import Observable from '../Observable';
 
 import {root} from '../util/root';
-import $iterator$ from '../util/Symbol_iterator';
+import $$iterator from '../util/Symbol_iterator';
 import tryCatch from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 
@@ -108,7 +108,7 @@ class StringIterator {
  private idx: number = 0,
  private len: number = str.length) {
  }
- [$iterator$]() { return (this); }
+ [$$iterator]() { return (this); }
  next() {
  return this.idx < this.len ? {
  done: false,
@@ -125,7 +125,7 @@ class ArrayIterator {
  private idx: number = 0,
  private len: number = toLength(arr)) {
  }
- [$iterator$]() { return this; }
+ [$$iterator]() { return this; }
  next() {
  return this.idx < this.len ? {
  done: false,
@@ -138,7 +138,7 @@ class ArrayIterator {
 }
 
 function getIterator(o) {
- const i = o[$iterator$];
+ const i = o[$$iterator];
  if (!i && typeof o === 'string') {
  return new StringIterator(o);
  }
@@ -148,7 +148,7 @@ function getIterator(o) {
  if (!i) {
  throw new TypeError('Object is not iterable');
  }
- return o[$iterator$]();
+ return o[$$iterator]();
 }
 
 function toLength(o) {

src/observables/PromiseObservable.ts

@@ -1,29 +1,42 @@
 import Observable from '../Observable';
 import Subscriber from '../Subscriber';
+import Scheduler from '../Scheduler';
+import Subscription from '../Subscription';
 
 export default class PromiseObservable<T> extends Observable<T> {
 
- static create<T>(promise: Promise<T>) {
- return new PromiseObservable(promise);
+ static create<T>(promise: Promise<T>, scheduler: Scheduler = Scheduler.immediate) {
+ return new PromiseObservable(promise, scheduler);
  }
 
- constructor(protected promise: Promise<T>) {
+ constructor(private promise: Promise<T>, private scheduler: Scheduler) {
  super();
  }
 
  _subscribe(subscriber: Subscriber<T>) {
- this.promise.then(
- (x) => {
- if(!subscriber.isUnsubscribed) {
- subscriber.next(x);
+ const scheduler = this.scheduler;
+ const promise = this.promise;
+
+ if (scheduler === Scheduler.immediate) {
+ promise.then(value => {
+ subscriber.next(value);
  subscriber.complete();
- }
- },
-  (e) => {
-  if(!subscriber.isUnsubscribed) {
-  subscriber.error(e);
-  }
- }
- );
+ },
+  err => subscriber.error(err));
+ } else {
+ let subscription = new Subscription();
+ promise.then(value => subscription.add(scheduler.schedule(0, { value, subscriber }, dispatchNext)),
+ err => subscription.add(scheduler.schedule(0, { err, subscriber }, dispatchError)));
+ return subscription;
+ }
  }
 }
+
+function dispatchNext({ value, subscriber }) {
+ subscriber.next(value);
+ subscriber.complete();
+}
+
+function dispatchError({ err, subscriber }) {
+ subscriber.error(err);
+}

src/util/Symbol_iterator.ts

@@ -16,7 +16,7 @@ if (!root.Symbol.iterator) {
  }
 }
 
-export default root.Symbol.dispose;
+export default root.Symbol.iterator;
 
 // // Shim in iterator support
 // export var $iterator$ = (typeof Symbol === 'function' && Symbol.iterator) || '_es6shim_iterator_';

src/util/isArray.ts

@@ -0,0 +1,3 @@
+const isArray = Array.isArray || (x => x && typeof x.length === 'number');
+
+export default isArray;

src/util/isObservable.ts

@@ -0,0 +1,5 @@
+import $$observer from './Symbol_observer';
+
+export default function isObservable(x): boolean {
+ return x && typeof x[$$observer] === 'function';
+}

src/util/isPromise.ts

@@ -0,0 +1,3 @@
+export default function isPromise(x): boolean {
+ return x && typeof x.then === 'function';
+}