Add support for react/promise v3

With v3 come a whole list of [changes](https://github.com/reactphp/promise/releases/tag/v3.0.0)
including end of promise chain detection, the removal of the `CancellablePromiseInterface`
interface, and type templating.

Author: WyriHaximus

.github/workflows/ci.yml

@@ -29,3 +29,27 @@ jobs:
           coverage: xdebug
       - run: composer install
       - run: vendor/bin/phpunit --coverage-text
+  PHPStan:
+    name: PHPStan (PHP ${{ matrix.php }} on ${{ matrix.os }})
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os:
+          - ubuntu-20.04
+          - windows-2019
+        php:
+          - 8.2
+          - 8.1
+          - 8.0
+          - 7.4
+          - 7.3
+          - 7.2
+    steps:
+      - uses: actions/checkout@v3
+      - uses: shivammathur/setup-php@v2
+        with:
+          php-version: ${{ matrix.php }}
+          coverage: none
+      - run: composer install
+      - run: vendor/bin/phpstan

composer.json

@@ -21,12 +21,13 @@
     ],
     "require": {
         "php": ">=7.0.0",
-        "react/promise": "~2.2"
+        "react/promise": "^3 || ~2.2"
     },
     "require-dev": {
         "satooshi/php-coveralls": "~1.0",
         "phpunit/phpunit": "^8.5 || ^9",
-        "react/event-loop": "^1.0 || ^0.5 || ^0.4.2"
+        "react/event-loop": "^1.0 || ^0.5 || ^0.4.2",
+        "phpstan/phpstan": "^1.10"
     },
     "suggest": {
         "react/event-loop": "Used for scheduling async operations"

phpstan.neon.dist

@@ -0,0 +1,8 @@
+parameters:
+    level: max
+
+    paths:
+        - test/types/
+
+    fileExtensions:
+        - php

src/Observable.php

@@ -78,6 +78,10 @@
 use Rx\Subject\ReplaySubject;
 use Rx\Subject\Subject;
 
+/**
+ * @template T
+ * @template-implements ObservableInterface<T>
+ */
 abstract class Observable implements ObservableInterface
 {
     /**
@@ -296,25 +300,25 @@ public function mergeAll(): Observable
     /**
      * Converts an array to an observable sequence
      *
-     * @param array $array
+     * @param array<T> $array
      * @param SchedulerInterface $scheduler
-     * @return ArrayObservable
+     * @return ObservableInterface<T>
      *
      * @demo fromArray/fromArray.php
      * @operator
      * @reactivex from
      */
-    public static function fromArray(array $array, SchedulerInterface $scheduler = null): ArrayObservable
+    public static function fromArray(array $array, SchedulerInterface $scheduler = null): ObservableInterface
     {
         return new ArrayObservable($array, $scheduler ?: Scheduler::getDefault());
     }
 
     /**
      * Converts an Iterator into an observable sequence
      *
-     * @param \Iterator $iterator
+     * @param \Iterator<T> $iterator
      * @param SchedulerInterface $scheduler
-     * @return IteratorObservable
+     * @return ObservableInterface<T>
      *
      * @demo iterator/iterator.php
      * @operator
@@ -2049,8 +2053,8 @@ public function finally(callable $callback): Observable
     /**
      * Converts a promise into an observable
      *
-     * @param PromiseInterface $promise
-     * @return Observable
+     * @param PromiseInterface<T> $promise
+     * @return Observable<T>
      * @throws \InvalidArgumentException
      *
      * @demo promise/fromPromise.php

src/Observable/ArrayObservable.php

@@ -6,15 +6,26 @@
 
 use Rx\DisposableInterface;
 use Rx\Observable;
+use Rx\ObservableInterface;
 use Rx\ObserverInterface;
 use Rx\SchedulerInterface;
 
+/**
+ * @template T
+ * @template-implements ObservableInterface<T>
+ */
 class ArrayObservable extends Observable
 {
+    /**
+     * @var array<T>
+     */
     private $data;
 
     private $scheduler;
 
+    /**
+     * @param array<T> $data
+     */
     public function __construct(array $data, SchedulerInterface $scheduler)
     {
         $this->data      = $data;

src/Observable/IteratorObservable.php

@@ -6,9 +6,14 @@
 
 use Rx\DisposableInterface;
 use Rx\Observable;
+use Rx\ObservableInterface;
 use Rx\ObserverInterface;
 use Rx\SchedulerInterface;
 
+/**
+ * @template T
+ * @template-implements ObservableInterface<T>
+ */
 class IteratorObservable extends Observable
 {
     private $items;

src/ObservableInterface.php

@@ -4,6 +4,9 @@
 
 namespace Rx;
 
+/**
+ * @template-covariant T
+ */
 interface ObservableInterface
 {
     /**

src/React/Promise.php

@@ -2,7 +2,6 @@
 
 namespace Rx\React;
 
-use React\Promise\CancellablePromiseInterface;
 use React\Promise\Promise as ReactPromise;
 use React\Promise\PromiseInterface;
 use Rx\Disposable\CallbackDisposable;
@@ -11,12 +10,16 @@
 use Rx\Observable\AnonymousObservable;
 use Rx\Subject\AsyncSubject;
 use React\Promise\Deferred;
+use Throwable;
 
+/**
+ * @template T
+ */
 final class Promise
 {
     /**
-     * @param mixed $value
-     * @return ReactPromise A promise resolved to $value
+     * @param T $value
+     * @return PromiseInterface<T> A promise resolved to $value
      */
     public static function resolved($value): ReactPromise
     {
@@ -27,21 +30,21 @@ public static function resolved($value): ReactPromise
 
     /**
      * @param mixed $exception
-     * @return ReactPromise A promise rejected with $exception
+     * @return PromiseInterface<never> A promise rejected with $exception
      */
     public static function rejected($exception): ReactPromise
     {
         $d = new Deferred();
-        $d->reject($exception);
+        $d->reject($exception instanceof Throwable ? $exception : new RejectedPromiseException($exception));
         return $d->promise();
     }
 
     /**
      * Converts an existing observable sequence to React Promise
      *
-     * @param ObservableInterface $observable
+     * @param ObservableInterface<T> $observable
      * @param Deferred $deferred
-     * @return ReactPromise
+     * @return ReactPromise<T>
      * @throws \InvalidArgumentException
      */
     public static function fromObservable(ObservableInterface $observable, Deferred $deferred = null): ReactPromise
@@ -94,7 +97,7 @@ function ($error) use ($subject) {
             $disp = $subject->subscribe($observer);
             return new CallbackDisposable(function () use ($p, $disp) {
                 $disp->dispose();
-                if ($p instanceof CancellablePromiseInterface) {
+                if (\method_exists($p, 'cancel')) {
                     $p->cancel();
                 }
             });

test/Rx/Functional/Promise/FromPromiseTest.php

@@ -40,7 +40,7 @@ function () {
      */
     public function from_promise_failure()
     {
-        $p = \React\Promise\reject('error');
+        $p = \React\Promise\reject(new RejectedPromiseException('error'));
 
         $source = Observable::fromPromise($p);
 

test/types/observable-from-array.php

@@ -0,0 +1,11 @@
+<?php
+
+use Rx\Observable;
+use Rx\Observable\ArrayObservable;
+use Rx\Scheduler\ImmediateScheduler;
+use function PHPStan\Testing\assertType;
+use function React\Promise\resolve;
+
+assertType('Rx\ObservableInterface<bool>', Observable::fromPromise(resolve(false)));
+assertType('Rx\Observable\ArrayObservable<bool>', Observable::fromPromise(resolve(false)));
+assertType('Rx\Observable<bool>', Observable::fromPromise(resolve(false)));