Do not cleanup subscription if next throws

spec/index.html

@@ -73,7 +73,6 @@ <h1>Subscription Observer Objects</h1>
   <ul>
     <li>If the observer's `error` method is called, the observer will not be invoked again and the observable's cleanup function will be called.</li>
     <li>If the observer's `complete` method is called, the observer will not be invoked again and the observable's cleanup function will be called.</li>
-    <li>If the observer throws an exception, the observable's cleanup function will be called.</li>
     <li>When the subscription is canceled, the observer will not be invoked again.</li>
   </ul>
 

spec/subscription-observer.html

@@ -41,15 +41,9 @@ <h1>%SubscriptionObserverPrototype%.next ( _value_ )</h1>
       1. If SubscriptionClosed(_subscription_) is *true*, return *undefined*.
       1. Let _observer_ be the value of _subscription_'s [[Observer]] internal slot.
       1. Assert: Type(_observer_) is Object.
-      1. Let _result_ be GetMethod(_observer_, `"next"`).
-      1. If _result_ is not an abrupt completion, then
-        1. Let _nextMethod_ be _result_.[[value]].
-        1. If _nextMethod_ is *undefined*, let _result_ be NormalCompletion(*undefined*).
-        1. Else, let _result_ be Call(_nextMethod_, _observer_, « ‍_value_ »).
-      1. If _result_ is an abrupt completion, then
-        1. Set _subscription_'s [[Observer]] internal slot to *undefined*.
-        1. Perform ? CleanupSubscription(_subscription_).
-      1. Return Completion(_result_).
+      1. Let _nextMethod_ be ? GetMethod(_observer_, `"next"`).
+      1. If _nextMethod_ is *undefined*, let _result_ be NormalCompletion(*undefined*).
+      1. Return ? Call(_nextMethod_, _observer_, « ‍_value_ »).
     </emu-alg>
   </emu-clause>
 

src/Observable.js

@@ -145,24 +145,14 @@ SubscriptionObserver.prototype = nonEnum({
             return undefined;
 
         let observer = subscription._observer;
+        let m = getMethod(observer, "next");
 
-        try {
-
-            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);
-
-        } catch (e) {
+        // If the observer doesn't support "next", then return undefined
+        if (!m)
+            return undefined;
 
-            // If the observer throws, then close the stream and rethrow the error
-            try { closeSubscription(subscription) }
-            finally { throw e }
-        }
+        // Send the next value to the sink
+        return m.call(observer, value);
     },
 
     error(value) {

test/observer-next.js

@@ -113,25 +113,14 @@ export default {
         });
 
         called = 0;
-        observable.subscribe({ next() { throw new Error() } });
+        let subscription = observable.subscribe({ next() { throw new Error() } });
         try { observer.next() }
         catch (x) {}
-        test._("Cleanup function is called when next throws an error")
-        .equals(called, 1);
-
-        let error = new Error(), caught = null;
-
-        new Observable(x => {
-            observer = x;
-            return _=> { throw new Error() };
-        }).subscribe({ next() { throw error } });
-
-        try { observer.next() }
-        catch (x) { caught = x }
+        test._("Cleanup function is not called when next throws an error")
+        .equals(called, 0);
 
-        test._("If both next and the cleanup function throw, then the error " +
-            "from the next method is thrown")
-        .assert(caught === error);
+        test._("Subscription is not closed when next throws an error")
+        .equals(subscription.closed, false);
 
     },