[WIP] Higher-Order Marble tests

src/Notification/OnNextNotification.php

@@ -4,6 +4,7 @@
 
 namespace Rx\Notification;
 
+use Rx\Observable;
 use Rx\ObserverInterface;
 use Rx\Notification;
 
@@ -30,6 +31,11 @@ protected function doAccept($onNext, $onError, $onCompleted)
 
     public function __toString(): string
     {
+        if ($this->value instanceof Observable) {
+            $messages = materializeObservable($this->value);
+            return '[' . implode(', ', $messages) . ']';
+        }
+
         return 'OnNext(' . json_encode($this->value) . ')';
     }
 

src/Testing/ColdObservable.php

@@ -26,14 +26,13 @@ public function __construct(TestScheduler $scheduler, array $messages = [])
 
     protected function _subscribe(ObserverInterface $observer): DisposableInterface
     {
+        $currentObservable     = $this;
+        $disposable            = new CompositeDisposable();
+        $scheduler             = $this->scheduler;
+        $isDisposed            = false;
         $this->subscriptions[] = new Subscription($this->scheduler->getClock());
         $index                 = count($this->subscriptions) - 1;
 
-        $currentObservable = $this;
-        $disposable        = new CompositeDisposable();
-        $scheduler         = $this->scheduler;
-        $isDisposed        = false;
-
         foreach ($this->messages as $message) {
             $notification = $message->getValue();
             $time         = $message->getTime();
@@ -53,7 +52,8 @@ protected function _subscribe(ObserverInterface $observer): DisposableInterface
         $subscriptions = &$this->subscriptions;
 
         return new CallbackDisposable(function () use (&$currentObservable, $index, $observer, $scheduler, &$subscriptions, &$isDisposed) {
-            $isDisposed            = true;
+            $isDisposed = true;
+
             $subscriptions[$index] = new Subscription($subscriptions[$index]->getSubscribed(), $scheduler->getClock());
         });
 

src/Testing/HotObservable.php

@@ -46,15 +46,12 @@ public function __construct(TestScheduler $scheduler, array $messages)
 
     protected function _subscribe(ObserverInterface $observer): DisposableInterface
     {
-        $currentObservable = $this;
-
+        $currentObservable     = $this;
         $this->observers[]     = $observer;
+        $subscriptions         = &$this->subscriptions;
         $this->subscriptions[] = new Subscription($this->scheduler->getClock());
-
-        $subscriptions = &$this->subscriptions;
-
-        $index     = count($this->subscriptions) - 1;
-        $scheduler = $this->scheduler;
+        $index                 = count($this->subscriptions) - 1;
+        $scheduler             = $this->scheduler;
 
         return new CallbackDisposable(function () use (&$currentObservable, $index, $observer, $scheduler, &$subscriptions) {
             $currentObservable->removeObserver($observer);

src/Testing/MockObserver.php

@@ -16,32 +16,34 @@ class MockObserver implements ObserverInterface
 {
     private $scheduler;
     private $messages = [];
+    private $startTime = 0;
 
-    public function __construct(TestScheduler $scheduler)
+    public function __construct(TestScheduler $scheduler, int $startTime = 0)
     {
         $this->scheduler = $scheduler;
+        $this->startTime = $startTime;
     }
 
     public function onNext($value)
     {
         $this->messages[] = new Recorded(
-            $this->scheduler->getClock(),
+            $this->scheduler->getClock() - $this->startTime,
             new OnNextNotification($value)
         );
     }
 
     public function onError(\Throwable $error)
     {
         $this->messages[] = new Recorded(
-            $this->scheduler->getClock(),
+            $this->scheduler->getClock() - $this->startTime,
             new OnErrorNotification($error)
         );
     }
 
     public function onCompleted()
     {
         $this->messages[] = new Recorded(
-            $this->scheduler->getClock(),
+            $this->scheduler->getClock() - $this->startTime,
             new OnCompletedNotification()
         );
     }

src/Testing/TestScheduler.php

@@ -7,7 +7,6 @@
 use Rx\Disposable\EmptyDisposable;
 use Rx\DisposableInterface;
 use Rx\ObserverInterface;
-use Rx\Scheduler;
 use Rx\Scheduler\VirtualTimeScheduler;
 
 class TestScheduler extends VirtualTimeScheduler

test/Rx/Functional/FunctionalTestCase.php

@@ -19,11 +19,19 @@ abstract class FunctionalTestCase extends TestCase
     /** @var  TestScheduler */
     protected $scheduler;
 
+    /** @var  TestScheduler */
+    static protected $globalScheduler;
+
     const TIME_FACTOR = 10;
 
     public function setup()
     {
         $this->scheduler = $this->createTestScheduler();
+        self::$globalScheduler = $this->scheduler;
+    }
+
+    static function getScheduler() {
+        return self::$globalScheduler;
     }
 
     /**
@@ -32,6 +40,8 @@ public function setup()
      */
     public function assertMessages(array $expected, array $recorded)
     {
+        $this->scheduler->start();
+
         if (count($expected) !== count($recorded)) {
             $this->fail(sprintf('Expected message count %d does not match actual count %d.', count($expected), count($recorded)));
         }
@@ -51,6 +61,8 @@ public function assertMessages(array $expected, array $recorded)
      */
     public function assertMessagesNotEqual(array $expected, array $recorded)
     {
+        $this->scheduler->start();
+
         if (count($expected) !== count($recorded)) {
             $this->assertTrue(true);
             return;
@@ -68,6 +80,8 @@ public function assertMessagesNotEqual(array $expected, array $recorded)
 
     public function assertSubscription(HotObservable $observable, Subscription $expected)
     {
+        $this->scheduler->start();
+
         $subscriptionCount = count($observable->getSubscriptions());
 
         if ($subscriptionCount === 0) {
@@ -89,6 +103,8 @@ public function assertSubscription(HotObservable $observable, Subscription $expe
 
     public function assertSubscriptions(array $expected, array $recorded)
     {
+        $this->scheduler->start();
+
         if (count($expected) !== count($recorded)) {
             $this->fail(sprintf('Expected subscription count %d does not match actual count %d.', count($expected), count($recorded)));
         }