Cherry picked higher-order observable test from v1

Author: davidwdan

src/Notification/OnNextObservableNotification.php

@@ -0,0 +1,56 @@
+<?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(Observable $value, TestScheduler $scheduler = null)
+    {
+        parent::__construct($value);
+        if (!$scheduler) {
+            $scheduler = new TestScheduler();
+        }
+
+        $this->observer = new MockHigherOrderObserver($scheduler, $scheduler->getClock());
+
+        $value->subscribe($this->observer);
+
+        $scheduler->start();
+    }
+
+    public function equals($other): bool
+    {
+        $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(): string
+    {
+        return '[' . implode(', ', $this->getMessages()) . ']';
+    }
+}

src/Testing/ColdObservable.php

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

src/Testing/HotObservable.php

@@ -48,17 +48,22 @@ protected function _subscribe(ObserverInterface $observer): DisposableInterface
     {
         $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());
+            }
         });
     }
 

src/Testing/MockHigherOrderObserver.php

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

src/Testing/MockObserver.php

@@ -7,6 +7,8 @@
 use Rx\Notification\OnCompletedNotification;
 use Rx\Notification\OnErrorNotification;
 use Rx\Notification\OnNextNotification;
+use Rx\Notification\OnNextObservableNotification;
+use Rx\Observable;
 use Rx\ObserverInterface;
 
 /**
@@ -16,32 +18,40 @@ 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)
     {
+        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(\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()
         );
     }