Merge pull request #149 from martinsik/2.x

Cherry picked distinct() optimization for 2.x

Author: davidwdan

src/Operator/DistinctOperator.php

@@ -22,13 +22,8 @@ final class DistinctOperator implements OperatorInterface
 
     public function __construct(callable $keySelector = null, callable $comparer = null)
     {
-
-        $this->comparer = $comparer ?: function ($x, $y) {
-            return $x == $y;
-        };
-
+        $this->comparer = $comparer;
         $this->keySelector = $keySelector;
-
     }
 
     public function __invoke(ObservableInterface $observable, ObserverInterface $observer): DisposableInterface
@@ -41,15 +36,22 @@ function ($value) use ($observer, &$values) {
                 try {
                     $key = $this->keySelector ? ($this->keySelector)($value) : $value;
 
-                    foreach ($values as $v) {
-                        $comparerEquals = ($this->comparer)($key, $v);
+                    if ($this->comparer) {
+                        foreach ($values as $v) {
+                            $comparerEquals = call_user_func($this->comparer, $key, $v);
 
-                        if ($comparerEquals) {
+                            if ($comparerEquals) {
+                                return;
+                            }
+                        }
+                        $values[] = $key;
+                    } else {
+                        if (array_key_exists($key, $values)) {
                             return;
                         }
+                        $values[$key] = null;
                     }
 
-                    $values[] = $key;
                     $observer->onNext($value);
 
                 } catch (\Throwable $e) {

test/Rx/Functional/Operator/DistinctTest.php

@@ -355,6 +355,45 @@ public function distinct_CustomKey_some_throw()
 
     }
 
+    /**
+     * @test
+     */
+    public function distinct_CustomKey_and_CustomComparer_some_duplicates()
+    {
+
+        $xs = $this->createHotObservable([
+            onNext(280, ['id' => 4]),
+            onNext(300, ['id' => 2]),
+            onNext(350, ['id' => 12]),
+            onNext(380, ['id' => 3]),
+            onNext(400, ['id' => 24]),
+            onCompleted(420)
+        ]);
+
+        $results = $this->scheduler->startWithCreate(function () use ($xs) {
+            return $xs->distinctKey(
+                function ($x) {
+                    return $x['id'];
+                },
+                $this->modComparer(10)
+            )->map(function ($x) {
+                return $x['id'];
+            });
+        });
+
+        $this->assertMessages([
+            onNext(280, 4),
+            onNext(300, 2),
+            onNext(380, 3),
+            onCompleted(420)
+        ], $results->getMessages());
+
+        $this->assertSubscriptions([
+            subscribe(200, 420)
+        ], $xs->getSubscriptions());
+
+    }
+
 
     public function modComparer($mod)
     {