feat(observable): add to/from Observable

spec/asynciterable-operators/toobservable-spec.ts

@@ -0,0 +1,67 @@
+import * as Ix from '../Ix';
+import * as test from 'tape';
+const { empty } = Ix.asynciterable;
+const { of } = Ix.asynciterable;
+const { _throw } = Ix.asynciterable;
+const { toObservable } = Ix.asynciterable;
+
+test('AsyncIterable#toObservable empty', async t => {
+  const xs = empty<number>();
+  const ys = toObservable(xs);
+  let fail = false;
+
+  ys.subscribe({
+    next: (value: number) => {
+      fail = true;
+    },
+    error: (err: any) => {
+      fail = true;
+    },
+    complete: () => {
+      t.false(fail);
+      t.end();
+    }
+  });
+});
+
+test('AsyncIterable#toObservable non-empty', async t => {
+  const results: number[] = [];
+  const xs = of(1, 2, 3);
+  const ys = toObservable(xs);
+  let fail = false;
+
+  ys.subscribe({
+    next: (value: number) => {
+      results.push(value);
+    },
+    error: (err: any) => {
+      fail = true;
+    },
+    complete: () => {
+      t.deepEqual(results, [1, 2, 3]);
+      t.false(fail);
+      t.end();
+    }
+  });
+});
+
+test('AsyncIterable#toObservable error', async t => {
+  const error = new Error();
+  const xs = _throw<number>(error);
+  const ys = toObservable(xs);
+  let fail = false;
+
+  ys.subscribe({
+    next: (value: number) => {
+      fail = true;
+    },
+    error: (err: any) => {
+      t.same(err, error);
+      t.false(fail);
+      t.end();
+    },
+    complete: () => {
+      fail = true;
+    }
+  });
+});

spec/asynciterable/fromobservable-spec.ts

@@ -0,0 +1,68 @@
+import * as Ix from '../Ix';
+import * as test from 'tape';
+const { fromObservable } = Ix.asynciterable;
+import { hasNext, noNext } from '../asynciterablehelpers';
+
+interface Observer<T> {
+  next: (value: T) => void;
+  error: (err: any) => void;
+  complete: () => void;
+}
+
+interface Subscription {
+  unsubscribe: () => void;
+}
+
+interface Observable<T> {
+  subscribe: (observer: Observer<T>) => Subscription;
+}
+
+class EmptySubscription implements Subscription {
+  unsubscribe() {
+    // tslint:disable-next-line:no-empty
+  }
+}
+
+class TestObservable<T> implements Observable<T> {
+  private _subscribe: (observer: Observer<T>) => Subscription;
+
+  constructor(subscribe: (observer: Observer<T>) => Subscription) {
+    this._subscribe = subscribe;
+  }
+
+  subscribe(observer: Observer<T>) {
+    return this._subscribe(observer);
+  }
+}
+
+test('AsyncIterable#fromObservable with completion', async t => {
+  const xs = new TestObservable<number>(obs => {
+    obs.next(42);
+    obs.complete();
+    return new EmptySubscription();
+  });
+  const ys = fromObservable(xs);
+
+  const it = ys[Symbol.asyncIterator]();
+  await hasNext(t, it, 42);
+  await noNext(t, it);
+  t.end();
+});
+
+test('AsyncIterable#fromObservable with error', async t => {
+  const err = new Error();
+  const xs = new TestObservable<number>(obs => {
+    obs.error(err);
+    return new EmptySubscription();
+  });
+  const ys = fromObservable(xs);
+
+  const it = ys[Symbol.asyncIterator]();
+  try {
+    await it.next();
+  } catch (e) {
+    t.same(err, e);
+  }
+
+  t.end();
+});

src/add/asynciterable-operators/toobservable.ts

@@ -0,0 +1,18 @@
+import { AsyncIterableX } from '../../asynciterable';
+import { toObservable } from '../../asynciterable/toobservable';
+import { Observable } from '../../observer';
+
+/**
+ * @ignore
+ */
+export function toObservableProto<TSource>(this: AsyncIterableX<TSource>): Observable<TSource> {
+  return toObservable(this);
+}
+
+AsyncIterableX.prototype.toObservable = toObservableProto;
+
+declare module '../../asynciterable' {
+  interface AsyncIterableX<T> {
+    toObservable: typeof toObservableProto;
+  }
+}

src/add/asynciterable/fromobservable.ts

@@ -0,0 +1,10 @@
+import { AsyncIterableX } from '../../asynciterable';
+import { fromObservable as fromObservableStatic } from '../../asynciterable/fromobservable';
+
+AsyncIterableX.fromObservable = fromObservableStatic;
+
+declare module '../../asynciterable' {
+  namespace AsyncIterableX {
+    export let fromObservable: typeof fromObservableStatic;
+  }
+}

src/asynciterable/__modules.ts

@@ -30,6 +30,7 @@ export { forEach } from './foreach';
 export { from } from './from';
 export { fromEvent } from './fromevent';
 export { fromEventPattern } from './fromeventpattern';
+export { fromObservable } from './fromobservable';
 export { generate } from './generate';
 export { generateTime } from './generatetime';
 export { groupBy } from './groupby';
@@ -86,6 +87,7 @@ export { throttle } from './throttle';
 export { _throw } from './throw';
 export { toArray } from './toarray';
 export { toMap } from './tomap';
+export { toObservable } from './toobservable';
 export { toSet } from './toset';
 export { union } from './union';
 export { _while } from './while';

src/asynciterable/toobservable.ts

@@ -37,6 +37,7 @@ class AsyncIterableObservable<TSource> implements Observable<TSource> {
           }
         });
     };
+    f();
 
     return subscription;
   }