override dispose() with process cleanup

Author: davidwdan

src/ProcessSubject.php

@@ -6,7 +6,6 @@
 use React\EventLoop\LoopInterface;
 use React\EventLoop\Timer\Timer;
 use Rx\Disposable\CallbackDisposable;
-use Rx\Disposable\EmptyDisposable;
 use Rx\ObserverInterface;
 use Rx\Subject\Subject;
 
@@ -46,7 +45,7 @@ public function subscribe(ObserverInterface $observer, $scheduler = null)
 
             try {
                 if ($this->process->isRunning()) {
-                    return new EmptyDisposable();
+                    return;
                 }
                 $this->process->start($timer->getLoop());
 
@@ -63,7 +62,7 @@ public function subscribe(ObserverInterface $observer, $scheduler = null)
                     parent::onCompleted();
                 });
             } catch (\Exception $e) {
-                $observer->onError($e);
+                parent::onError($e);
             }
         });
 
@@ -73,6 +72,12 @@ public function subscribe(ObserverInterface $observer, $scheduler = null)
         });
     }
 
+    public function dispose()
+    {
+        parent::dispose();
+        $this->process->terminate();
+    }
+
     /**
      * @return Process
      */