Added Scheduler to Promise::fromObservable() (#99)

* added Scheduler to Promise::fromObservable(), resolves #40 * fix docbock
ReactiveX · Dec 9, 2016 · de88548 · de88548
1 parent b93d296
commit de88548
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 3 deletions.
diff --git a/lib/Rx/React/Promise.php b/lib/Rx/React/Promise.php
@@ -8,6 +8,7 @@
 use Rx\Observable;
 use Rx\Observable\AnonymousObservable;
 use Rx\Observer\CallbackObserver;
+use Rx\SchedulerInterface;
 use Rx\Subject\AsyncSubject;
 use React\Promise\Deferred;
 
@@ -38,10 +39,12 @@ public static function rejected($exception)
     /**
      * Converts an existing observable sequence to React Promise
      *
-     * @param PromisorInterface|null $deferred
+     * @param ObservableInterface $observable
+     * @param \React\Promise\PromisorInterface|null $deferred
+     * @param SchedulerInterface|null $scheduler
      * @return \React\Promise\Promise
      */
-    public static function fromObservable(ObservableInterface $observable, Deferred $deferred = null)
+    public static function fromObservable(ObservableInterface $observable, Deferred $deferred = null, $scheduler = null)
     {
         $d     = $deferred ?: new Deferred();
         $value = null;
@@ -56,7 +59,7 @@ function ($error) use ($d) {
             function () use ($d, &$value) {
                 $d->resolve($value);
             }
-        ));
+        ), $scheduler);
 
         return $d->promise();
     }

diff --git a/test/Rx/Functional/React/PromiseFromObservableTest.php b/test/Rx/Functional/React/PromiseFromObservableTest.php
@@ -35,6 +35,33 @@ function () {
 
     }
 
+    /**
+     * @test
+     *
+     */
+    public function promise_success_scheduler()
+    {
+
+        $scheduler = $this->createTestScheduler();
+        $observer = $scheduler->createObserver();
+
+        $scheduler->scheduleAbsoluteWithState(null, 200, function($scheduler) use ($observer) {
+            $source = Observable::just(42)->delay(100);
+            $promise = Promise::fromObservable($source, null, $scheduler);
+
+            $promise->then(function($value) use ($observer) {
+                $observer->onNext($value);
+            });
+        });
+
+        $scheduler->start();
+
+        $this->assertMessages([
+            onNext(301, 42),
+        ], $observer->getMessages());
+
+    }
+
     /**
      * @test
      *