Add types

package.json

@@ -2,20 +2,26 @@
   "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/scheduler": "^1.2.2",
+    "@most/types": "^1.0.1",
+    "rollup": "^1.15.6",
+    "rollup-plugin-typescript2": "^0.21.1",
+    "typescript": "^3.5.2"
   }
 }

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" },
+    { file: r(__dirname, "dist/index.mjs"), format: "esm" }
+  ],
+  plugins: [
+    typescript({
+      typescript: require("typescript"),
+      cacheRoot: r(__dirname, "node_modules/.cache/rollup-plugin-typescript2")
+    })
+  ],
+  external: [
+    ...Object.keys(pkg.dependencies),
+    ...Object.keys(pkg.peerDependencies)
+  ]
+};

src/index.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 default fromObservable;

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;
+}

tsconfig.json

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