isEmpty operator 2.x branch

demo/bootstrap.php

@@ -19,6 +19,8 @@
 function asString($value) {
     if (is_array($value)) {
         return json_encode($value);
+    } elseif (is_bool($value)) {
+        return (string)(integer)$value;
     }
     return (string) $value;
 }

demo/isEmpty/isEmpty-false.php

@@ -0,0 +1,8 @@
+<?php
+
+require_once __DIR__ . '/../bootstrap.php';
+
+$source = \Rx\Observable::just(1)
+    ->isEmpty();
+
+$source->subscribe($stdoutObserver);

demo/isEmpty/isEmpty-false.php.expect

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

demo/isEmpty/isEmpty.php

@@ -0,0 +1,8 @@
+<?php
+
+require_once __DIR__ . '/../bootstrap.php';
+
+$source = \Rx\Observable::emptyObservable()
+    ->isEmpty();
+
+$source->subscribe($stdoutObserver);

demo/isEmpty/isEmpty.php.expect

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

src/Observable.php

@@ -36,6 +36,7 @@
 use Rx\Operator\DistinctUntilChangedOperator;
 use Rx\Operator\DoOnEachOperator;
 use Rx\Operator\GroupByUntilOperator;
+use Rx\Operator\IsEmptyOperator;
 use Rx\Operator\MapOperator;
 use Rx\Operator\FilterOperator;
 use Rx\Operator\MinOperator;
@@ -1922,6 +1923,23 @@ public function throttle(int $throttleDuration, SchedulerInterface $scheduler =
         });
     }
 
+    /**
+     * If the source Observable is empty it returns an Observable that emits true, otherwise it emits false.
+     *
+     * @return Observable
+     *
+     * @demo isEmpty/isEmpty.php
+     * @demo isEmpty/isEmpty-false.php
+     * @operator
+     * @reactivex contains
+     */
+    public function isEmpty(): Observable
+    {
+        return $this->lift(function () {
+            return new IsEmptyOperator();
+        });
+    }
+
     /**
      * @param Promise $promise
      * @param SchedulerInterface|null $scheduler

src/Operator/IsEmptyOperator.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace Rx\Operator;
+
+use Rx\DisposableInterface;
+use Rx\ObservableInterface;
+use Rx\ObserverInterface;
+
+final class IsEmptyOperator implements OperatorInterface
+{
+
+    public function __invoke(ObservableInterface $observable, ObserverInterface $observer): DisposableInterface
+    {
+        return $observable->subscribe(
+            function() use ($observer) {
+                $observer->onNext(false);
+                $observer->onCompleted();
+            },
+            [$observer, 'onError'],
+            function() use ($observer) {
+                $observer->onNext(true);
+                $observer->onCompleted();
+            }
+        );
+    }
+
+}

test/Rx/Functional/Operator/IsEmptyTest.php

@@ -0,0 +1,194 @@
+<?php
+
+namespace Rx\Functional\Operator;
+
+use Rx\Functional\FunctionalTestCase;
+
+class IsEmptyTest extends FunctionalTestCase
+{
+
+    /**
+     * @test
+     */
+    public function should_return_true_if_source_is_empty()
+    {
+        $xs = $this->createHotObservable([
+            onCompleted(300)
+        ]);
+
+        $results = $this->scheduler->startWithCreate(function() use ($xs) {
+            return $xs->isEmpty();
+        });
+
+        $this->assertMessages([
+            onNext(300, true),
+            onCompleted(300),
+        ], $results->getMessages());
+
+        $this->assertSubscriptions([
+            subscribe(200, 300),
+        ], $xs->getSubscriptions());
+    }
+
+    /**
+     * @test
+     */
+    public function should_return_false_if_source_emits_element()
+    {
+        $xs = $this->createHotObservable([
+            onNext(150, 'a'),
+            onNext(300, 'b'),
+            onCompleted(300)
+        ]);
+
+        $results = $this->scheduler->startWithCreate(function() use ($xs) {
+            return $xs->isEmpty();
+        });
+
+        $this->assertMessages([
+            onNext(300, false),
+            onCompleted(300),
+        ], $results->getMessages());
+
+        $this->assertSubscriptions([
+            subscribe(200, 300),
+        ], $xs->getSubscriptions());
+    }
+
+    /**
+     * @test
+     */
+    public function should_return_true_if_source_emits_before_subscription()
+    {
+        $xs = $this->createHotObservable([
+            onNext(150, 'a'),
+            onCompleted(300)
+        ]);
+
+        $results = $this->scheduler->startWithCreate(function() use ($xs) {
+            return $xs->isEmpty();
+        });
+
+        $this->assertMessages([
+            onNext(300, true),
+            onCompleted(300),
+        ], $results->getMessages());
+
+        $this->assertSubscriptions([
+            subscribe(200, 300),
+        ], $xs->getSubscriptions());
+    }
+
+    /**
+     * @test
+     */
+    public function should_raise_error_if_source_raise_error()
+    {
+        $e = new \Exception();
+
+        $xs = $this->createHotObservable([
+            onError(300, $e),
+        ]);
+
+        $results = $this->scheduler->startWithCreate(function() use ($xs) {
+            return $xs->isEmpty();
+        });
+
+        $this->assertMessages([
+            onError(300, $e),
+        ], $results->getMessages());
+
+        $this->assertSubscriptions([
+            subscribe(200, 300),
+        ], $xs->getSubscriptions());
+    }
+
+    /**
+     * @test
+     */
+    public function should_not_complete_if_source_never_emits()
+    {
+        $xs = $this->createHotObservable([]);
+
+        $results = $this->scheduler->startWithCreate(function() use ($xs) {
+            return $xs->isEmpty();
+        });
+
+        $this->assertMessages([], $results->getMessages());
+        $this->assertSubscriptions([
+            subscribe(200, 1000),
+        ], $xs->getSubscriptions());
+    }
+
+    /**
+     * @test
+     */
+    public function should_return_true_if_source_completes_immediately()
+    {
+        $xs = $this->createHotObservable([
+            onCompleted(201),
+        ]);
+
+        $results = $this->scheduler->startWithCreate(function() use ($xs) {
+            return $xs->isEmpty();
+        });
+
+        $this->assertMessages([
+            onNext(201, true),
+            onCompleted(201),
+        ], $results->getMessages());
+        $this->assertSubscriptions([
+            subscribe(200, 201),
+        ], $xs->getSubscriptions());
+    }
+
+    /**
+     * @test
+     */
+    public function should_allow_unsubscribing_explicitly_and_early()
+    {
+        $xs = $this->createHotObservable([
+            onNext(600, 'a'),
+            onNext(700, 'b'),
+        ]);
+
+        $results = $this->scheduler->startWithDispose(function() use ($xs) {
+            return $xs->isEmpty();
+        }, 500);
+
+        $this->assertMessages([], $results->getMessages());
+
+        $this->assertSubscriptions([
+            subscribe(200, 500),
+        ], $xs->getSubscriptions());
+    }
+
+    /**
+     * @test
+     */
+    public function should_not_break_unsubscription_chains_when_result_is_unsubscribed_explicitly()
+    {
+        $xs = $this->createHotObservable([
+            onNext(600, 'a'),
+            onNext(700, 'b'),
+        ]);
+
+        $results = $this->scheduler->startWithDispose(function() use ($xs) {
+            return $xs
+                ->flatMap(function($value) {
+                    return \Rx\Observable::just($value);
+                })
+                ->isEmpty()
+                ->flatMap(function($value) {
+                    return \Rx\Observable::just($value);
+                });
+        }, 500);
+
+        $this->assertMessages([], $results->getMessages());
+
+        $this->assertSubscriptions([
+            subscribe(200, 500),
+        ], $xs->getSubscriptions());
+    }
+
+}