The finally() operator for 2.x branch (#132)

* added finally operator

* finally() operator for 2.x

Author: martinsik

demo/finally/finally-error.php

@@ -0,0 +1,15 @@
+<?php
+
+require_once __DIR__ . '/../bootstrap.php';
+
+Rx\Observable::range(1, 3)
+    ->map(function($value) {
+        if ($value == 2) {
+            throw new \Exception('error');
+        }
+        return $value;
+    })
+    ->finally(function() {
+        echo "Finally\n";
+    })
+    ->subscribe($stdoutObserver);

demo/finally/finally-error.php.expect

@@ -0,0 +1,3 @@
+Next value: 1
+Exception: error
+Finally

demo/finally/finally.php

@@ -0,0 +1,9 @@
+<?php
+
+require_once __DIR__ . '/../bootstrap.php';
+
+Rx\Observable::range(1, 3)
+    ->finally(function() {
+        echo "Finally\n";
+    })
+    ->subscribe($stdoutObserver);

demo/finally/finally.php.expect

@@ -0,0 +1,5 @@
+Next value: 1
+Next value: 2
+Next value: 3
+Complete!
+Finally

src/Observable.php

@@ -35,6 +35,7 @@
 use Rx\Operator\DistinctOperator;
 use Rx\Operator\DistinctUntilChangedOperator;
 use Rx\Operator\DoOnEachOperator;
+use Rx\Operator\FinallyOperator;
 use Rx\Operator\GroupByUntilOperator;
 use Rx\Operator\IsEmptyOperator;
 use Rx\Operator\MapOperator;
@@ -1940,6 +1941,24 @@ public function isEmpty(): Observable
         });
     }
 
+    /**
+     * Will call a specified function when the source terminates on complete or error.
+     *
+     * @param callable $callback
+     * @return Observable
+     *
+     * @demo finally/finally.php
+     * @demo finally/finally-error.php
+     * @operator
+     * @reactivex do
+     */
+    public function finally(callable $callback): Observable
+    {
+        return $this->lift(function() use ($callback) {
+            return new FinallyOperator($callback);
+        });
+    }
+
     /**
      * @param Promise $promise
      * @param SchedulerInterface|null $scheduler

src/Operator/FinallyOperator.php

@@ -0,0 +1,36 @@
+<?php
+
+namespace Rx\Operator;
+
+use Rx\DisposableInterface;
+use Rx\Disposable\CallbackDisposable;
+use Rx\Disposable\BinaryDisposable;
+use Rx\ObservableInterface;
+use Rx\ObserverInterface;
+
+class FinallyOperator implements OperatorInterface
+{
+    /** @var callable */
+    private $callback;
+
+    /**
+     * @param callable $callback
+     */
+    public function __construct(callable $callback)
+    {
+        $this->callback = $callback;
+    }
+
+    /**
+     * @param \Rx\ObservableInterface $observable
+     * @param \Rx\ObserverInterface $observer
+     * @return \Rx\DisposableInterface
+     */
+    public function __invoke(ObservableInterface $observable, ObserverInterface $observer): DisposableInterface
+    {
+        return new BinaryDisposable(
+            $observable->subscribe($observer),
+            new CallbackDisposable($this->callback)
+        );
+    }
+}