min operator (#64)

Author: Громов Андрей Олегович

demo/min/min-with-comparer.php

@@ -0,0 +1,18 @@
+<?php
+
+require_once __DIR__ . "/../bootstrap.php";
+
+/* With a comparer */
+$comparer = function ($x, $y) {
+    if ($x > $y) {
+        return 1;
+    } elseif ($x < $y) {
+        return -1;
+    }
+    return 0;
+};
+
+$source = \Rx\Observable::fromArray([1, 3, 5, 7, 9, 2, 4, 6, 8])
+    ->min($comparer);
+
+$subscription = $source->subscribe($createStdoutObserver());

demo/min/min-with-comparer.php.expect

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

demo/min/min.php

@@ -0,0 +1,9 @@
+<?php
+
+require_once __DIR__ . "/../bootstrap.php";
+
+/* Without comparer */
+$source = \Rx\Observable::fromArray([1, 3, 5, 7, 9, 2, 4, 6, 8])
+    ->min();
+
+$subscription = $source->subscribe($createStdoutObserver());

demo/min/min.php.expect

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

lib/Rx/Observable.php

@@ -33,6 +33,7 @@
 use Rx\Operator\GroupByUntilOperator;
 use Rx\Operator\MapOperator;
 use Rx\Operator\FilterOperator;
+use Rx\Operator\MinOperator;
 use Rx\Operator\MaxOperator;
 use Rx\Operator\MaterializeOperator;
 use Rx\Operator\MergeAllOperator;
@@ -1505,7 +1506,24 @@ public function startWithArray(array $startArray)
             return new StartWithArrayOperator($startArray);
         });
     }
-
+    
+    /**
+     * Returns the minimum value in an observable sequence according to the specified comparer.
+     *
+     * @param callable $comparer
+     * @return AnonymousObservable
+     *
+     * @demo min/min.php
+     * @demo min/min-with-comparer.php
+     * @operator
+     * @reactivex min
+     */
+    public function min(callable $comparer = null){
+        return $this->lift(function () use ($comparer) {
+            return new MinOperator($comparer);
+        });
+    }
+    
     /**
      * Returns the maximum value in an observable sequence according to the specified comparer.
      *

lib/Rx/Operator/MinOperator.php

@@ -0,0 +1,71 @@
+<?php
+
+namespace Rx\Operator;
+
+use Rx\ObservableInterface;
+use Rx\Observer\CallbackObserver;
+use Rx\ObserverInterface;
+use Rx\SchedulerInterface;
+
+class MinOperator implements OperatorInterface
+{
+    /** @var callable|null */
+    private $comparer;
+
+    /**
+     * MinOperator constructor.
+     * @param $comparer callable
+     */
+    public function __construct(callable $comparer = null)
+    {
+        if ($comparer === null) {
+            $comparer = function ($x, $y) {
+                return $x > $y ? 1 : ($x < $y ? -1 : 0);
+            };
+        }
+
+        $this->comparer = $comparer;
+    }
+
+    /**
+     * @inheritDoc
+     */
+    public function __invoke(
+        ObservableInterface $observable,
+        ObserverInterface $observer,
+        SchedulerInterface $scheduler = null
+    ) {
+        $previousMin = null;
+        $comparing   = false;
+
+        return $observable->subscribe(new CallbackObserver(
+            function ($x) use (&$comparing, &$previousMin, $observer) {
+                if (!$comparing) {
+                    $comparing = true;
+                    $previousMin = $x;
+
+                    return;
+                }
+
+                try {
+                    $result = call_user_func($this->comparer, $x, $previousMin);
+                    if ($result < 0) {
+                        $previousMin = $x;
+                    }
+                } catch (\Exception $e) {
+                    $observer->onError($e);
+                }
+            },
+            [$observer, 'onError'],
+            function () use (&$comparing, &$previousMin, $observer) {
+                if ($comparing) {
+                    $observer->onNext($previousMin);
+                    $observer->onCompleted();
+                    return;
+                }
+
+                $observer->onError(new \Exception("Empty"));
+            }
+            ), $scheduler);
+    }
+}