Add support template types

.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,12 @@
+parameters:
+    level: max
+
+    paths:
+        - src/Observable.php
+        - src/ObservableInterface.php
+        - src/Observable/
+        - src/Subject/
+        - test/types/
+
+    fileExtensions:
+        - php

src/Observable.php

@@ -78,10 +78,13 @@
 use Rx\Subject\ReplaySubject;
 use Rx\Subject\Subject;
 
+/**
+ * @template-covariant T
+ */
 abstract class Observable implements ObservableInterface
 {
     /**
-     * @param callable|ObserverInterface|null $onNextOrObserver
+     * @param callable(T)|ObserverInterface|null $onNextOrObserver
      * @param callable|null $onError
      * @param callable|null $onCompleted
      * @return DisposableInterface
@@ -175,15 +178,15 @@ public static function interval(int $interval, AsyncSchedulerInterface $schedule
     /**
      * Returns an observable sequence that contains a single element.
      *
-     * @param mixed $value Single element in the resulting observable sequence.
-     * @param SchedulerInterface $scheduler
-     * @return ReturnObservable An observable sequence with the single element.
+     * @param T $value Single element in the resulting observable sequence.
+     * @param ?SchedulerInterface $scheduler
+     * @return Observable<T>
      *
      * @demo of/of.php
      * @operator
      * @reactivex just
      */
-    public static function of($value, SchedulerInterface $scheduler = null): ReturnObservable
+    public static function of($value, SchedulerInterface $scheduler = null): Observable
     {
         return new ReturnObservable($value, $scheduler ?: Scheduler::getDefault());
     }
@@ -194,9 +197,9 @@ public static function of($value, SchedulerInterface $scheduler = null): ReturnO
      *
      * @param $value
      * @param SchedulerInterface|null $scheduler
-     * @return ReturnObservable
+     * @return Observable
      */
-    public static function just($value, SchedulerInterface $scheduler = null): ReturnObservable
+    public static function just($value, SchedulerInterface $scheduler = null): Observable
     {
         return static::of($value, $scheduler);
     }
@@ -296,25 +299,25 @@ public function mergeAll(): Observable
     /**
      * Converts an array to an observable sequence
      *
-     * @param array $array
-     * @param SchedulerInterface $scheduler
-     * @return ArrayObservable
+     * @param array<T> $array
+     * @param ?SchedulerInterface $scheduler
+     * @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 +2052,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/AnonymousObservable.php

@@ -7,11 +7,19 @@
 use Rx\Disposable\CallbackDisposable;
 use Rx\DisposableInterface;
 use Rx\Observable;
+use Rx\ObservableInterface;
 use Rx\ObserverInterface;
 use Rx\Observer\AutoDetachObserver;
 
+/**
+ * @template T
+ * @template-extends Observable<T>
+ */
 class AnonymousObservable extends Observable
 {
+    /**
+     * @var callable
+     */
     private $subscribeAction;
 
     public function __construct(callable $subscribeAction)

src/Observable/ArrayObservable.php

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

src/Observable/ConnectableObservable.php

@@ -12,27 +12,29 @@
 use Rx\Subject\Subject;
 
 /**
+ * @template T
+ * @template-extends Observable<T>
  * Class ConnectableObservable
  * @package Rx\Observable
  */
 class ConnectableObservable extends Observable
 {
-    /** @var \Rx\Subject\Subject */
+    /** @var \Rx\Subject\Subject<T> */
     protected $subject;
 
     /** @var  BinaryDisposable */
     protected $subscription;
 
-    /** @var  Observable */
+    /** @var  Observable<T> */
     protected $sourceObservable;
 
     /** @var bool */
     protected $hasSubscription;
 
     /**
      * ConnectableObservable constructor.
-     * @param Observable $source
-     * @param \Rx\Subject\Subject $subject
+     * @param Observable<T> $source
+     * @param \Rx\Subject\Subject<T> $subject
      */
     public function __construct(Observable $source, Subject $subject = null)
     {
@@ -63,6 +65,9 @@ public function connect(): DisposableInterface
         return $this->subscription;
     }
 
+    /**
+     * @return RefCountObservable<T>
+     */
     public function refCount(): RefCountObservable
     {
         return new RefCountObservable($this);

src/Observable/EmptyObservable.php

@@ -9,8 +9,15 @@
 use Rx\ObserverInterface;
 use Rx\SchedulerInterface;
 
+/**
+ * @template T
+ * @template-extends Observable<T>
+ */
 class EmptyObservable extends Observable
 {
+    /**
+     * @var SchedulerInterface
+     */
     private $scheduler;
 
     public function __construct(SchedulerInterface $scheduler)

src/Observable/ErrorObservable.php

@@ -9,9 +9,20 @@
 use Rx\ObserverInterface;
 use Rx\SchedulerInterface;
 
+/**
+ * @template T
+ * @template-extends Observable<T>
+ */
 class ErrorObservable extends Observable
 {
+    /**
+     * @var \Throwable
+     */
     private $error;
+
+    /**
+     * @var SchedulerInterface
+     */
     private $scheduler;
 
     public function __construct(\Throwable $error, SchedulerInterface $scheduler)

src/Observable/ForkJoinObservable.php

@@ -10,19 +10,35 @@
 use Rx\ObserverInterface;
 use Rx\Disposable\CompositeDisposable;
 
+/**
+ * @template T
+ * @template-extends Observable<T>
+ */
 class ForkJoinObservable extends Observable
 {
     /**
-     * @var Observable[]
+     * @var array<Observable<T>>
      */
     private $observables;
 
+    /**
+     * @var array<T>
+     */
     private $values = [];
 
+    /**
+     * @var int
+     */
     private $completed = 0;
 
+    /**
+     * @var callable|null
+     */
     private $resultSelector;
 
+    /**
+     * @param array<Observable<T>> $observables
+     */
     public function __construct(array $observables = [], callable $resultSelector = null)
     {
         $this->observables    = $observables;

src/Observable/GroupedObservable.php

@@ -11,11 +11,26 @@
 use Rx\Disposable\RefCountDisposable;
 use Rx\DisposableInterface;
 
+/**
+ * @template T
+ * @template-extends Observable<T>
+ */
 class GroupedObservable extends Observable
 {
+    /**
+     * @var mixed
+     */
     private $key;
+
+    /**
+     * @var ObservableInterface<T>
+     */
     private $underlyingObservable;
 
+    /**
+     * @param mixed $key
+     * @param ObservableInterface<T> $underlyingObservable
+     */
     public function __construct($key, ObservableInterface $underlyingObservable, RefCountDisposable $mergedDisposable = null)
     {
         $this->key = $key;
@@ -25,6 +40,9 @@ public function __construct($key, ObservableInterface $underlyingObservable, Ref
             $this->newUnderlyingObservable($mergedDisposable, $underlyingObservable);
     }
 
+    /**
+     * @return mixed
+     */
     public function getKey()
     {
         return $this->key;
@@ -35,6 +53,10 @@ protected function _subscribe(ObserverInterface $observer): DisposableInterface
         return $this->underlyingObservable->subscribe($observer);
     }
 
+    /**
+     * @param ObservableInterface<T> $underlyingObservable
+     * @return Observable<T>
+     */
     private function newUnderlyingObservable(RefCountDisposable $mergedDisposable, ObservableInterface $underlyingObservable): Observable
     {
         return new AnonymousObservable(

src/Observable/IntervalObservable.php

@@ -9,8 +9,15 @@
 use Rx\ObserverInterface;
 use Rx\AsyncSchedulerInterface;
 
+/**
+ * @template T
+ * @template-extends Observable<T>
+ */
 class IntervalObservable extends Observable
 {
+    /**
+     * @var int
+     */
     private $interval;
 
     /** @var AsyncSchedulerInterface */