BREAKING: Align with current TC39 proposal

Author: Kevin Smith

.gitignore

@@ -1,3 +1,3 @@
 .DS_Store
-_*
 node_modules
+zen-observable.js

.npmignore

@@ -0,0 +1,2 @@
+.DS_Store
+node_modules

LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2015 zenparsing (Kevin Smith)
+Copyright (c) 2018 zenparsing (Kevin Smith)
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -180,20 +180,3 @@ Observable.of(0, 1, 2, 3, 4).reduce((previousValue, currentValue) => {
 ```
 
 Returns a new Observable that applies a function against an accumulator and each value of the stream to reduce it to a single value.
-
-
-### observable.flatMap ( callback )
-
-Returns a new Observable that emits the values from each Observable that is returned from the `callback` argument.
-
-```js
-Observable.of("Hello", "Goodbye").flatMap(value => {
-    return Observable.of(value + " Earth", value + " Mars");
-}).subscribe(value => {
-    console.log(value);
-});
-// "Hello Earth"
-// "Hello Mars"
-// "Goodbye Earth"
-// "Goodbye Mars"
-```

package-lock.json

@@ -1,14 +1,18 @@
 {
   "name": "zen-observable",
-  "version": "0.6.1",
+  "version": "0.7.0",
   "repository": "zenparsing/zen-observable",
   "description": "An Implementation of ES Observables",
   "homepage": "https://github.com/zenparsing/zen-observable",
   "license": "MIT",
   "devDependencies": {
-    "es-observable-tests": "^0.3.0",
-    "moon-unit": "^0.2.1",
-    "v8-promise": "*"
+    "esdown": "^1.2.8",
+    "moon-unit": "^0.2.1"
   },
-  "dependencies": {}
+  "dependencies": {},
+  "scripts": {
+    "test": "esdown test",
+    "build": "esdown - src/Observable.js zen-observable.js -g '*'",
+    "prepublishOnly": "npm run build"
+  }
 }

package.json

@@ -15,6 +15,10 @@ if (typeof Symbol === "function" && !Symbol.observable) {
 
 // === Abstract Operations ===
 
+function hostReportError(e) {
+  setTimeout(() => { throw e });
+}
+
 function getMethod(obj, key) {
   let value = obj[key];
 
@@ -59,7 +63,8 @@ function cleanupSubscription(subscription) {
   subscription._cleanup = undefined;
 
   // Call the cleanup function
-  cleanup();
+  try { cleanup() }
+  catch (e) { hostReportError(e) }
 }
 
 function subscriptionClosed(subscription) {
@@ -88,10 +93,12 @@ function Subscription(observer, subscriber) {
   this._cleanup = undefined;
   this._observer = observer;
 
-  let start = getMethod(observer, "start");
-
-  if (start)
-    start.call(observer, this);
+  try {
+    let start = getMethod(observer, "start");
+    if (start) start.call(observer, this);
+  } catch (e) {
+    hostReportError(e);
+  }
 
   if (subscriptionClosed(this))
     return;
@@ -141,73 +148,68 @@ addMethods(SubscriptionObserver.prototype = {}, {
 
     // If the stream is closed, then return undefined
     if (subscriptionClosed(subscription))
-      return undefined;
+      return;
 
     let observer = subscription._observer;
-    let m = getMethod(observer, "next");
 
-    // If the observer doesn't support "next", then return undefined
-    if (!m)
-      return undefined;
-
-    // Send the next value to the sink
-    return m.call(observer, value);
+    try {
+      // If the observer has a "next" method, send the next value
+      let m = getMethod(observer, "next");
+      if (m) m.call(observer, value);
+    } catch (e) {
+      hostReportError(e);
+    }
   },
 
   error(value) {
     let subscription = this._subscription;
 
     // If the stream is closed, throw the error to the caller
-    if (subscriptionClosed(subscription))
-      throw value;
+    if (subscriptionClosed(subscription)) {
+      hostReportError(value);
+      return;
+    }
 
     let observer = subscription._observer;
     subscription._observer = undefined;
 
     try {
       let m = getMethod(observer, "error");
-
-      // If the sink does not support "error", then throw the error to the caller
-      if (!m)
-        throw value;
-
-      value = m.call(observer, value);
+      if (m) m.call(observer, value);
+      else throw value;
     } catch (e) {
-      try { cleanupSubscription(subscription) }
-      finally { throw e }
+      hostReportError(e);
     }
 
     cleanupSubscription(subscription);
-    return value;
   },
 
-  complete(value) {
+  complete() {
     let subscription = this._subscription;
 
-    // If the stream is closed, then return undefined
     if (subscriptionClosed(subscription))
-      return undefined;
+      return;
 
     let observer = subscription._observer;
     subscription._observer = undefined;
 
     try {
       let m = getMethod(observer, "complete");
-
-      // If the sink does not support "complete", then return undefined
-      value = m ? m.call(observer, value) : undefined;
+      if (m) m.call(observer);
     } catch (e) {
-      try { cleanupSubscription(subscription) }
-      finally { throw e }
+      hostReportError(e);
     }
 
     cleanupSubscription(subscription);
-    return value;
   },
 
 });
 
 export function Observable(subscriber) {
+  // Constructor cannot be called as a function
+  if (!(this instanceof Observable))
+    throw new TypeError("Observable cannot be called as a function");
+
   // The stream subscriber must be a function
   if (typeof subscriber !== "function")
     throw new TypeError("Observable initializer must be a function");
@@ -224,6 +226,8 @@ addMethods(Observable.prototype, {
         error: args[0],
         complete: args[1],
       };
+    } else if (typeof observer !== 'object' || observer === null) {
+      observer = {};
     }
 
     return new Subscription(observer, this._subscriber);
@@ -251,7 +255,7 @@ addMethods(Observable.prototype, {
             return;
 
           try {
-            return fn(value);
+            fn(value);
           } catch (err) {
             reject(err);
             subscription.unsubscribe();
@@ -278,11 +282,11 @@ addMethods(Observable.prototype, {
         try { value = fn(value) }
         catch (e) { return observer.error(e) }
 
-        return observer.next(value);
+        observer.next(value);
       },
 
-      error(e) { return observer.error(e) },
-      complete(x) { return observer.complete(x) },
+      error(e) { observer.error(e) },
+      complete() { observer.complete() },
     }));
   },
 
@@ -297,14 +301,14 @@ addMethods(Observable.prototype, {
         if (observer.closed)
           return;
 
-        try { if (!fn(value)) return undefined }
+        try { if (!fn(value)) return }
         catch (e) { return observer.error(e) }
 
-        return observer.next(value);
+        observer.next(value);
       },
 
-      error(e) { return observer.error(e) },
-      complete() { return observer.complete() },
+      error(e) { observer.error(e) },
+      complete() { observer.complete() },
     }));
   },
 
@@ -339,8 +343,7 @@ addMethods(Observable.prototype, {
 
       complete() {
         if (!hasValue && !hasSeed) {
-          observer.error(new TypeError("Cannot reduce an empty sequence"));
-          return;
+          return observer.error(new TypeError("Cannot reduce an empty sequence"));
         }
 
         observer.next(acc);
@@ -350,70 +353,6 @@ addMethods(Observable.prototype, {
     }));
   },
 
-  flatMap(fn) {
-    if (typeof fn !== "function")
-      throw new TypeError(fn + " is not a function");
-
-    let C = getSpecies(this);
-
-    return new C(observer => {
-      let completed = false;
-      let subscriptions = [];
-
-      // Subscribe to the outer Observable
-      let outer = this.subscribe({
-
-        next(value) {
-          if (fn) {
-            try {
-              value = fn(value);
-            } catch (x) {
-              observer.error(x);
-              return;
-            }
-          }
-
-          // Subscribe to the inner Observable
-          Observable.from(value).subscribe({
-            _subscription: null,
-
-            start(s) { subscriptions.push(this._subscription = s) },
-            next(value) { observer.next(value) },
-            error(e) { observer.error(e) },
-
-            complete() {
-              let i = subscriptions.indexOf(this._subscription);
-
-              if (i >= 0)
-                subscriptions.splice(i, 1);
-
-              closeIfDone();
-            }
-          });
-        },
-
-        error(e) {
-          return observer.error(e);
-        },
-
-        complete() {
-          completed = true;
-          closeIfDone();
-        }
-      });
-
-      function closeIfDone() {
-        if (completed && subscriptions.length === 0)
-          observer.complete();
-      }
-
-      return () => {
-        subscriptions.forEach(s => s.unsubscribe());
-        outer.unsubscribe();
-      };
-    });
-  },
-
 });
 
 Object.defineProperty(Observable.prototype, getSymbol("observable"), {

src/Observable.js

@@ -1,17 +1,18 @@
-import { runTests } from "es-observable-tests";
+import { runTests } from "./tc39";
 import { Observable } from "../src/Observable.js";
 import { TestRunner } from "moon-unit";
 
-import flatMapTests from "./flatMap.js";
 import reduceTests from "./reduce.js";
 import mapTests from "./map.js";
 import filterTests from "./filter.js";
 import speciesTests from "./species.js";
 
+// Silence setTimeout so errors aren't thrown
+global.setTimeout = function() {};
+
 runTests(Observable).then(() => {
   return new TestRunner().inject({ Observable }).run({
     "map": mapTests,
-    "flatMap": flatMapTests,
     "reduce": reduceTests,
     "filter": filterTests,
     "species": speciesTests,