Testing higher-order Observables (#144)

* enable testing higher-order observables

* refactored into onNext

* remove redundant argument in OnNextObservableNotification

* cleanup

* make sure automatic subscriptions aren't logged

Author: martinsik

lib/Rx/Notification/OnNextObservableNotification.php

@@ -0,0 +1,57 @@
+<?php
+
+namespace Rx\Notification;
+
+use Rx\Observable;
+use Rx\Testing\MockHigherOrderObserver;
+use Rx\Testing\TestScheduler;
+
+class OnNextObservableNotification extends OnNextNotification
+{
+    /** @var MockHigherOrderObserver */
+    private $observer;
+
+    public function __construct($value, TestScheduler $scheduler = null)
+    {
+        parent::__construct($value);
+        if (!$scheduler) {
+            $scheduler = new TestScheduler();
+        }
+
+        $this->observer = new MockHigherOrderObserver($scheduler, $scheduler->getClock());
+
+        /** @var Observable $value */
+        $value->subscribe($this->observer, $scheduler);
+
+        $scheduler->start();
+    }
+
+    public function equals($other)
+    {
+        $messages1 = $this->getMessages();
+        /** @var OnNextObservableNotification $other */
+        $messages2 = $other->getMessages();
+
+        if (count($messages1) != count($messages2)) {
+            return false;
+        }
+
+        for ($i = 0; $i < count($messages1); $i++) {
+            if (!$messages1[$i]->equals($messages2[$i])) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+    public function getMessages()
+    {
+        return $this->observer->getMessages();
+    }
+
+    public function __toString()
+    {
+        return '[' . join(', ', $this->getMessages()) . ']';
+    }
+}

lib/Rx/Testing/ColdObservable.php

@@ -22,13 +22,16 @@ public function __construct($scheduler, $messages = [])
 
     public function subscribe(ObserverInterface $observer, $scheduler = null)
     {
-        $this->subscriptions[] = new Subscription($this->scheduler->getClock());
-        $index                 = count($this->subscriptions) - 1;
-
+        $scheduler         = $scheduler ?: $this->scheduler;
         $currentObservable = $this;
         $disposable        = new CompositeDisposable();
-        $scheduler         = $this->scheduler;
         $isDisposed        = false;
+        $index             = null;
+
+        if (!($observer instanceof MockHigherOrderObserver)) {
+            $this->subscriptions[] = new Subscription($scheduler->getClock());
+            $index = count($this->subscriptions) - 1;
+        }
 
         foreach ($this->messages as $message) {
             $notification = $message->getValue();
@@ -50,7 +53,9 @@ public function subscribe(ObserverInterface $observer, $scheduler = null)
 
         return new CallbackDisposable(function () use (&$currentObservable, $index, $observer, $scheduler, &$subscriptions, &$isDisposed) {
             $isDisposed            = true;
-            $subscriptions[$index] = new Subscription($subscriptions[$index]->getSubscribed(), $scheduler->getClock());
+            if (!($observer instanceof MockHigherOrderObserver)) {
+                $subscriptions[$index] = new Subscription($subscriptions[$index]->getSubscribed(), $scheduler->getClock());
+            }
         });
 
     }

lib/Rx/Testing/HotObservable.php

@@ -43,17 +43,22 @@ public function subscribe(ObserverInterface $observer, $scheduler = null)
     {
         $currentObservable = $this;
 
-        $this->observers[]     = $observer;
-        $this->subscriptions[] = new Subscription($this->scheduler->getClock());
+        $this->observers[] = $observer;
+        $subscriptions     = &$this->subscriptions;
+        $index             = null;
 
-        $subscriptions = &$this->subscriptions;
+        if (!($observer instanceof MockHigherOrderObserver)) {
+            $this->subscriptions[] = new Subscription($this->scheduler->getClock());
+            $index = count($this->subscriptions) - 1;
+        }
 
-        $index     = count($this->subscriptions) - 1;
         $scheduler = $this->scheduler;
 
         return new CallbackDisposable(function () use (&$currentObservable, $index, $observer, $scheduler, &$subscriptions) {
             $currentObservable->removeObserver($observer);
-            $subscriptions[$index] = new Subscription($subscriptions[$index]->getSubscribed(), $scheduler->getClock());
+            if (!($observer instanceof MockHigherOrderObserver)) {
+                $subscriptions[$index] = new Subscription($subscriptions[$index]->getSubscribed(), $scheduler->getClock());
+            }
         });
     }
 

lib/Rx/Testing/MockHigherOrderObserver.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace Rx\Testing;
+
+class MockHigherOrderObserver extends MockObserver
+{
+}

lib/Rx/Testing/MockObserver.php

@@ -6,41 +6,55 @@
 use Rx\Notification\OnCompletedNotification;
 use Rx\Notification\OnErrorNotification;
 use Rx\Notification\OnNextNotification;
+use Rx\Notification\OnNextObservableNotification;
+use Rx\Observable;
 use Rx\ObserverInterface;
 
 /**
  * Mock observer that records all messages.
  */
 class MockObserver implements ObserverInterface
 {
+    /** @var TestScheduler */
     private $scheduler;
+
+    /** @var Recorded[] */
     private $messages = [];
 
-    public function __construct($scheduler)
+    private $startTime = 0;
+
+    public function __construct($scheduler, $startTime = 0)
     {
         $this->scheduler = $scheduler;
+        $this->startTime = $startTime;
     }
 
     public function onNext($value)
     {
+        if ($value instanceof Observable) {
+            $notification = new OnNextObservableNotification($value, $this->scheduler);
+        } else {
+            $notification = new OnNextNotification($value);
+        }
+
         $this->messages[] = new Recorded(
-            $this->scheduler->getClock(),
-            new OnNextNotification($value)
+            $this->scheduler->getClock() - $this->startTime,
+            $notification
         );
     }
 
     public function onError(Exception $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()
         );
     }