The finally() operator

demo/do/doFinally-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;
+    })
+    ->doFinally(function() {
+        echo "Finally\n";
+    })
+    ->subscribe($stdoutObserver);

demo/do/doFinally-error.php.expect

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

demo/do/doFinally.php

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

demo/do/doFinally.php.expect

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

lib/Rx/Observable.php

@@ -32,6 +32,7 @@
 use Rx\Operator\DistinctOperator;
 use Rx\Operator\DistinctUntilChangedOperator;
 use Rx\Operator\DoOnEachOperator;
+use Rx\Operator\DoFinallyOperator;
 use Rx\Operator\GroupByUntilOperator;
 use Rx\Operator\IsEmptyOperator;
 use Rx\Operator\MapOperator;
@@ -1867,4 +1868,23 @@ public function isEmpty()
             return new IsEmptyOperator();
         });
     }
+
+
+    /**
+     * Will call a specified function when the source terminates on complete or error.
+     *
+     * @param callable $callback
+     * @return AnonymousObservable
+     *
+     * @demo do/doFinally.php
+     * @demo do/doFinally-error.php
+     * @operator
+     * @reactivex do
+     */
+    public function doFinally(callable $callback)
+    {
+        return $this->lift(function() use ($callback) {
+            return new DoFinallyOperator($callback);
+        });
+    }
 }

lib/Rx/Operator/DoFinallyOperator.php

@@ -0,0 +1,37 @@
+<?php
+
+namespace Rx\Operator;
+
+use Rx\Disposable\CallbackDisposable;
+use Rx\Disposable\BinaryDisposable;
+use Rx\ObservableInterface;
+use Rx\ObserverInterface;
+use Rx\SchedulerInterface;
+
+class DoFinallyOperator 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
+     * @param \Rx\SchedulerInterface $scheduler
+     * @return \Rx\DisposableInterface
+     */
+    public function __invoke(ObservableInterface $observable, ObserverInterface $observer, SchedulerInterface $scheduler = null)
+    {
+        return new BinaryDisposable(
+            $observable->subscribe($observer, $scheduler),
+            new CallbackDisposable($this->callback)
+        );
+    }
+}