Reroute exceptions in onNext to onError by default (#193)

* Send exceptions in onNext to error handler by default

* Verify proper disposal on onNext throw defaulting to onError

Author: mbonneau

src/Observable.php

@@ -6,6 +6,7 @@
 
 use React\Promise\Deferred;
 use React\Promise\PromiseInterface;
+use Rx\Disposable\EmptyDisposable;
 use Rx\Observable\AnonymousObservable;
 use Rx\Observable\ArrayObservable;
 use Rx\Observable\ConnectableObservable;
@@ -97,9 +98,26 @@ public function subscribe($onNextOrObserver = null, callable $onError = null, ca
             throw new \InvalidArgumentException('The first argument needs to be a "callable" or "Observer"');
         }
 
-        $observer = new CallbackObserver($onNextOrObserver, $onError, $onCompleted);
+        $disposable = new EmptyDisposable();
+
+        $observer = new CallbackObserver(
+            $onNextOrObserver === null
+                ? null
+                : function ($value) use ($onNextOrObserver, &$observer, &$disposable) {
+                try {
+                    $onNextOrObserver($value);
+                } catch (\Throwable $throwable) {
+                    $disposable->dispose();
+                    $observer->onError($throwable);
+                }
+            },
+            $onError,
+            $onCompleted
+        );
+
+        $disposable = $this->_subscribe($observer);
 
-        return $this->_subscribe($observer);
+        return $disposable;
     }
 
     /**

test/Rx/Functional/ObservableTest.php

@@ -0,0 +1,53 @@
+<?php
+
+namespace Rx\Functional;
+
+use Rx\Observer\TestException;
+use Rx\Testing\MockObserver;
+
+class ObservableTest extends FunctionalTestCase
+{
+    public function testExceptionInOnNextByDefaultGoesToErrorAndDisposes()
+    {
+        $xs = $this->createHotObservable(
+            [
+                onNext(150, 1),
+                onNext(210, 2),
+                onNext(220, 3),
+                onNext(230, 4),
+                onCompleted(250)
+            ]);
+
+        $results = new MockObserver($this->scheduler);
+
+        $disposable = null;
+
+        $this->scheduler->scheduleAbsolute(200, function () use ($xs, $results, &$disposable) {
+            $disposable = $xs->subscribe(
+                function ($value) use ($results) {
+                    if ($value === 3) {
+                        throw new TestException();
+                    }
+                    $results->onNext($value);
+                },
+                [$results, 'onError'],
+                [$results, 'onCompleted']
+            );
+        });
+
+        $this->scheduler->scheduleAbsolute(1000, function () use (&$disposable) {
+            $disposable->dispose();
+        });
+
+        $this->scheduler->start();
+
+        $this->assertMessages(
+            [
+                onNext(210, 2),
+                onError(220, new TestException()),
+            ],
+            $results->getMessages());
+
+        $this->assertSubscriptions([subscribe(200, 220)], $xs->getSubscriptions());
+    }
+}

test/Rx/ObservableTest.php

@@ -8,6 +8,7 @@
 use Rx\Observable\EmptyObservable;
 use Rx\Observable\RefCountObservable;
 use Rx\Observable\ReturnObservable;
+use Rx\Observer\TestException;
 use Rx\Scheduler\ImmediateScheduler;
 use Rx\Testing\TestScheduler;
 
@@ -179,4 +180,26 @@ public function testSwitchLatestCallsSwitch()
 
         $o->switchLatest();
     }
+
+    /**
+     * @test
+     */
+    public function it_sends_throwables_in_onnext_to_onerror()
+    {
+        $onNext = function ($x) {
+            throw new TestException();
+        };
+
+        $error = null;
+
+        Observable::of(0)
+            ->subscribe(
+                $onNext,
+                function (\Throwable $e) use (&$error) {
+                    $error = $e;
+                }
+            );
+
+        $this->assertInstanceOf(TestException::class, $error);
+    }
 }