Add toObservable function

package.json

@@ -2,20 +2,27 @@
   "name": "most-observable",
   "version": "0.0.1",
   "description": "Create @most/core compatible streams from observables",
-  "main": "dist/main.js",
+  "main": "dist/index.js",
+  "module": "dist/index.mjs",
+  "types": "dist/index.d.ts",
   "author": "Kanit Sharma",
   "license": "MIT",
   "scripts": {
-    "build": "node build.js"
+    "build": "rollup -c"
   },
   "dependencies": {
     "symbol-observable": "^1.2.0"
   },
   "peerDependencies": {
-    "@most/core": "^1.3.2"
+    "@most/core": "^1.3.2",
+    "@most/scheduler": "^1.2.2"
   },
   "devDependencies": {
-    "@lectro/core": "^1.1.2-alpha-21",
-    "@lectro/enhancer-buildutils": "^0.1.0-alpha-7"
+    "@most/core": "^1.3.2",
+    "@most/scheduler": "^1.2.2",
+    "@most/types": "^1.0.1",
+    "rollup": "^1.15.6",
+    "rollup-plugin-typescript2": "^0.21.1",
+    "typescript": "^3.5.2"
   }
 }

readme.md

@@ -1,6 +1,6 @@
 # most-observable
 
-Create `@most/core` streams from [ES observables](https://github.com/tc39/proposal-observable).
+Convert `@most/core` streams to/from [ES observables](https://github.com/tc39/proposal-observable).
 
 ## Install
 
@@ -13,7 +13,7 @@ Create `@most/core` streams from [ES observables](https://github.com/tc39/propos
 Creating state stream from redux store as it is an Observable but any [ES observable](https://github.com/tc39/proposal-observable) should work.
 
 ```javascript
-import fromObservable from "most-observable";
+import { fromObservable } from "most-observable";
 import { runEffects, tap } from "@most/core";
 import { newDefaultScheduler } from "@most/scheduler";
 
@@ -31,3 +31,9 @@ runEffects(tap(console.log, state$), newDefaultScheduler());
 ```javascript
 const stream = fromObservable(observable);
 ```
+
+- toObservable
+
+```javascript
+const observable = toObservable(stream);
+```

rollup.config.js

@@ -0,0 +1,21 @@
+import { resolve as r } from "path";
+import typescript from "rollup-plugin-typescript2";
+import * as pkg from "./package.json";
+
+export default {
+  input: r(__dirname, "src/index.ts"),
+  output: [
+    { file: r(__dirname, "dist/index.js"), format: "cjs", exports: "named" },
+    { file: r(__dirname, "dist/index.mjs"), format: "esm", exports: "named" }
+  ],
+  plugins: [
+    typescript({
+      typescript: require("typescript"),
+      cacheRoot: r(__dirname, "node_modules/.cache/rollup-plugin-typescript2")
+    })
+  ],
+  external: [
+    ...Object.keys(pkg.dependencies),
+    ...Object.keys(pkg.peerDependencies)
+  ]
+};

src/fromObservable.ts

@@ -0,0 +1,51 @@
+import { currentTime } from "@most/scheduler";
+import { Scheduler, Sink, Stream, Time } from "@most/types";
+import observableSymbol from "symbol-observable";
+import { ObservableCompatible, ObservableLike } from "./observableTypes";
+
+const fromObservable = <T>(observable: ObservableCompatible<T>): Stream<T> =>
+  new ObservableStream(observable);
+
+const tryEvent = <T>(t: Time, x: T, sink: Sink<T>) => {
+  try {
+    sink.event(t, x);
+  } catch (e) {
+    sink.error(t, e);
+  }
+};
+
+const getObservable = <T>(o?: ObservableCompatible<T>) => {
+  if (o) {
+    const method = (o as any)[observableSymbol] as
+      | (() => ObservableLike<T>)
+      | undefined;
+
+    if (typeof method === "function") {
+      const obs = method.call(o);
+      if (!(obs && typeof obs.subscribe === "function")) {
+        throw new TypeError("invalid observable " + obs);
+      }
+      return obs;
+    }
+  }
+};
+
+class ObservableStream<T> implements Stream<T> {
+  constructor(protected readonly observable: ObservableCompatible<T>) {}
+
+  run(sink: Sink<T>, scheduler: Scheduler) {
+    const send = (e: T) => tryEvent(currentTime(scheduler), e, sink);
+
+    const subscription = getObservable(this.observable)!.subscribe({
+      next: send,
+      error: e => sink.error(currentTime(scheduler), e),
+      complete: () => sink.end(currentTime(scheduler))
+    });
+
+    const dispose = () => subscription.unsubscribe();
+
+    return { dispose };
+  }
+}
+
+export { fromObservable };

src/index.ts

@@ -0,0 +1,2 @@
+export { fromObservable as default, fromObservable } from "./fromObservable";
+export { toObservable } from "./toObservable";

src/observableTypes.ts

@@ -0,0 +1,19 @@
+// types ported from "https://github.com/bigslycat/es-observable"
+
+export interface ObservableCompatible<T> {
+  [Symbol.observable](): ObservableLike<T>;
+}
+
+export interface ObserverLike<T> {
+  next: (value: T) => void;
+  error: (errorValue: Error) => void;
+  complete: () => void;
+}
+
+export interface ObservableLike<T> extends ObservableCompatible<T> {
+  subscribe: (observer: Partial<ObserverLike<T>>) => SubscriptionLike;
+}
+
+export interface SubscriptionLike {
+  unsubscribe(): void;
+}

src/toObservable.ts

@@ -0,0 +1,72 @@
+import { run } from "@most/core";
+import { Disposable, Scheduler, Sink, Stream, Time } from "@most/types";
+import observableSymbol from "symbol-observable";
+import {
+  ObservableLike,
+  ObserverLike,
+  SubscriptionLike
+} from "./observableTypes";
+
+const toObservable = <T>(
+  stream: Stream<T>,
+  scheduler: Scheduler
+): ObservableLike<T> => {
+  return new StreamObservable(stream, scheduler);
+};
+
+class StreamObservable<T> implements ObservableLike<T> {
+  constructor(
+    protected readonly stream: Stream<T>,
+    protected readonly scheduler: Scheduler
+  ) {}
+
+  [Symbol.observable](): this;
+  [observableSymbol]() {
+    return this;
+  }
+
+  subscribe(observer: Partial<ObserverLike<T>>) {
+    const disposable = run(
+      new ObserverSink(observer),
+      this.scheduler,
+      this.stream
+    );
+    return new StreamSubscription(disposable);
+  }
+}
+
+class StreamSubscription implements SubscriptionLike {
+  constructor(protected readonly disposable: Disposable) {}
+
+  unsubscribe() {
+    this.disposable.dispose();
+  }
+}
+
+const noop = () => {};
+
+class ObserverSink<T> implements Sink<T> {
+  constructor(protected readonly observer: Partial<ObserverLike<T>>) {}
+
+  obsNext = this.observer.next ? this.observer.next.bind(this.observer) : noop;
+  obsError = this.observer.error
+    ? this.observer.error.bind(this.observer)
+    : noop;
+  obsComplete = this.observer.complete
+    ? this.observer.complete.bind(this.observer)
+    : noop;
+
+  event(_time: Time, v: T) {
+    this.obsNext(v);
+  }
+
+  error(_time: Time, e: Error) {
+    this.obsError(e);
+  }
+
+  end(_time: Time) {
+    this.obsComplete();
+  }
+}
+
+export { toObservable };

tsconfig.json

@@ -0,0 +1,11 @@
+{
+  "compilerOptions": {
+    "target": "esnext",
+    "module": "esnext",
+    "moduleResolution": "node",
+    "allowSyntheticDefaultImports": true,
+    "strict": true,
+    "declaration": true
+  },
+  "include": ["src/**/*"]
+}