Improve type definitions and update to PHPStan level max

README.md

@@ -111,7 +111,7 @@ from this source stream.
 The event receives a single mixed argument for incoming data.
 
 ```php
-$stream->on('data', function ($data) {
+$stream->on('data', function (mixed $data): void {
     echo $data;
 });
 ```
@@ -142,7 +142,7 @@ The `end` event will be emitted once the source stream has successfully
 reached the end of the stream (EOF).
 
 ```php
-$stream->on('end', function () {
+$stream->on('end', function (): void {
     echo 'END';
 });
 ```
@@ -180,7 +180,7 @@ trying to read from this stream.
 The event receives a single `Exception` argument for the error instance.
 
 ```php
-$server->on('error', function (Exception $e) {
+$server->on('error', function (Exception $e): void {
     echo 'Error: ' . $e->getMessage() . PHP_EOL;
 });
 ```
@@ -213,7 +213,7 @@ stream which should result in the same error processing.
 The `close` event will be emitted once the stream closes (terminates).
 
 ```php
-$stream->on('close', function () {
+$stream->on('close', function (): void {
     echo 'CLOSED';
 });
 ```
@@ -312,7 +312,7 @@ Re-attach the data source after a previous `pause()`.
 ```php
 $stream->pause();
 
-Loop::addTimer(1.0, function () use ($stream) {
+Loop::addTimer(1.0, function () use ($stream): void {
     $stream->resume();
 });
 ```
@@ -362,7 +362,7 @@ you'll have to manually close the destination stream:
 
 ```php
 $source->pipe($dest);
-$source->on('close', function () use ($dest) {
+$source->on('close', function () use ($dest): void {
     $dest->end('BYE!');
 });
 ```
@@ -456,7 +456,7 @@ The `drain` event will be emitted whenever the write buffer became full
 previously and is now ready to accept more data.
 
 ```php
-$stream->on('drain', function () use ($stream) {
+$stream->on('drain', function () use ($stream): void {
     echo 'Stream is now ready to accept more data';
 });
 ```
@@ -478,11 +478,11 @@ The event receives a single `ReadableStreamInterface` argument for the
 source stream.
 
 ```php
-$stream->on('pipe', function (ReadableStreamInterface $source) use ($stream) {
+$stream->on('pipe', function (ReadableStreamInterface $source) use ($stream): void {
     echo 'Now receiving piped data';
 
     // explicitly close target if source emits an error
-    $source->on('error', function () use ($stream) {
+    $source->on('error', function () use ($stream): void {
         $stream->close();
     });
 });
@@ -506,7 +506,7 @@ trying to write to this stream.
 The event receives a single `Exception` argument for the error instance.
 
 ```php
-$stream->on('error', function (Exception $e) {
+$stream->on('error', function (Exception $e): void {
     echo 'Error: ' . $e->getMessage() . PHP_EOL;
 });
 ```
@@ -536,7 +536,7 @@ stream which should result in the same error processing.
 The `close` event will be emitted once the stream closes (terminates).
 
 ```php
-$stream->on('close', function () {
+$stream->on('close', function (): void {
     echo 'CLOSED';
 });
 ```
@@ -746,7 +746,7 @@ stream in order to stop waiting for the stream to flush its final data.
 
 ```php
 $stream->end();
-Loop::addTimer(1.0, function () use ($stream) {
+Loop::addTimer(1.0, function () use ($stream): void {
     $stream->close();
 });
 ```
@@ -831,10 +831,10 @@ readable mode or a stream such as `STDIN`:
 
 ```php
 $stream = new ReadableResourceStream(STDIN);
-$stream->on('data', function ($chunk) {
+$stream->on('data', function (string $chunk): void {
     echo $chunk;
 });
-$stream->on('end', function () {
+$stream->on('end', function (): void {
     echo 'END';
 });
 ```
@@ -1121,7 +1121,7 @@ used to convert data, for example for transforming any structured data into
 a newline-delimited JSON (NDJSON) stream like this:
 
 ```php
-$through = new ThroughStream(function ($data) {
+$through = new ThroughStream(function (mixed $data): string {
     return json_encode($data) . PHP_EOL;
 });
 $through->on('data', $this->expectCallableOnceWith("[2, true]\n"));
@@ -1133,7 +1133,7 @@ The callback function is allowed to throw an `Exception`. In this case,
 the stream will emit an `error` event and then [`close()`](#close-1) the stream.
 
 ```php
-$through = new ThroughStream(function ($data) {
+$through = new ThroughStream(function (mixed $data): string {
     if (!is_string($data)) {
         throw new \UnexpectedValueException('Only strings allowed');
     }
@@ -1164,7 +1164,7 @@ $stdout = new WritableResourceStream(STDOUT);
 
 $stdio = new CompositeStream($stdin, $stdout);
 
-$stdio->on('data', function ($chunk) use ($stdio) {
+$stdio->on('data', function (string $chunk) use ($stdio): void {
     $stdio->write('You said: ' . $chunk);
 });
 ```
@@ -1243,7 +1243,7 @@ If you do not want to run these, they can simply be skipped like this:
 vendor/bin/phpunit --exclude-group internet
 ```
 
-On top of this, we use PHPStan on level 5 to ensure type safety across the project:
+On top of this, we use PHPStan on max level to ensure type safety across the project:
 
 ```bash
 vendor/bin/phpstan

examples/01-http.php

@@ -26,10 +26,10 @@
 
 $stream = new DuplexResourceStream($resource);
 
-$stream->on('data', function ($chunk) {
+$stream->on('data', function (string $chunk): void {
     echo $chunk;
 });
-$stream->on('close', function () {
+$stream->on('close', function (): void {
     echo '[CLOSED]' . PHP_EOL;
 });
 

examples/02-https.php

@@ -26,10 +26,10 @@
 
 $stream = new DuplexResourceStream($resource);
 
-$stream->on('data', function ($chunk) {
+$stream->on('data', function (string $chunk): void {
     echo $chunk;
 });
-$stream->on('close', function () {
+$stream->on('close', function (): void {
     echo '[CLOSED]' . PHP_EOL;
 });
 

examples/91-benchmark-throughput.php

@@ -22,8 +22,11 @@
 
 $args = getopt('i:o:t:');
 $if = $args['i'] ?? '/dev/zero';
+assert(is_string($if));
 $of = $args['o'] ?? '/dev/null';
+assert(is_string($of));
 $t  = $args['t'] ?? 1;
+assert(is_numeric($t));
 
 // passing file descriptors requires mapping paths (https://bugs.php.net/bug.php?id=53465)
 $if = str_replace('/dev/fd/', 'php://fd/', $if);
@@ -38,18 +41,21 @@
 
 // setup input and output streams and pipe inbetween
 $fh = fopen($if, 'r');
+assert(is_resource($fh));
+$fo = fopen($of, 'w');
+assert(is_resource($fo));
 $in = new React\Stream\ReadableResourceStream($fh);
-$out = new React\Stream\WritableResourceStream(fopen($of, 'w'));
+$out = new React\Stream\WritableResourceStream($fo);
 $in->pipe($out);
 
 // stop input stream in $t seconds
 $start = microtime(true);
-$timeout = Loop::addTimer($t, function () use ($in) {
+$timeout = Loop::addTimer((float) $t, function () use ($in): void {
     $in->close();
 });
 
 // print stream position once stream closes
-$in->on('close', function () use ($fh, $start, $timeout, $info) {
+$in->on('close', function () use ($fh, $start, $timeout, $info): void {
     $t = microtime(true) - $start;
     Loop::cancelTimer($timeout);
 

phpstan.neon.dist

@@ -1,5 +1,5 @@
 parameters:
-    level: 5
+    level: max
 
     paths:
         - examples/

src/CompositeStream.php

@@ -6,8 +6,13 @@
 
 final class CompositeStream extends EventEmitter implements DuplexStreamInterface
 {
+    /** @var ReadableStreamInterface */
     private $readable;
+
+    /** @var WritableStreamInterface */
     private $writable;
+
+    /** @var bool */
     private $closed = false;
 
     public function __construct(ReadableStreamInterface $readable, WritableStreamInterface $writable)

src/DuplexResourceStream.php

@@ -9,6 +9,7 @@
 
 final class DuplexResourceStream extends EventEmitter implements DuplexStreamInterface
 {
+    /** @var resource */
     private $stream;
 
     /** @var LoopInterface */
@@ -31,11 +32,20 @@ final class DuplexResourceStream extends EventEmitter implements DuplexStreamInt
      * @var int
      */
     private $bufferSize;
+
+    /** @var WritableStreamInterface */
     private $buffer;
 
+    /** @var bool */
     private $readable = true;
+
+    /** @var bool */
     private $writable = true;
+
+    /** @var bool */
     private $closing = false;
+
+    /** @var bool */
     private $listening = false;
 
     /**
@@ -78,13 +88,13 @@ public function __construct($stream, ?LoopInterface $loop = null, ?int $readChun
         $this->bufferSize = $readChunkSize ?? 65536;
         $this->buffer = $buffer;
 
-        $this->buffer->on('error', function ($error) {
+        $this->buffer->on('error', function (\Exception $error): void {
             $this->emit('error', [$error]);
         });
 
         $this->buffer->on('close', [$this, 'close']);
 
-        $this->buffer->on('drain', function () {
+        $this->buffer->on('drain', function (): void {
             $this->emit('drain');
         });
 
@@ -167,11 +177,14 @@ public function pipe(WritableStreamInterface $dest, array $options = []): Writab
         return Util::pipe($this, $dest, $options);
     }
 
-    /** @internal */
-    public function handleData($stream)
+    /**
+     * @internal
+     * @param resource $stream
+     */
+    public function handleData($stream): void
     {
         $error = null;
-        \set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error): bool {
+        \set_error_handler(function (int $errno, string $errstr, string $errfile, int $errline) use (&$error): bool {
             $error = new \ErrorException(
                 $errstr,
                 0,

src/ReadableResourceStream.php

@@ -37,7 +37,10 @@ final class ReadableResourceStream extends EventEmitter implements ReadableStrea
      */
     private $bufferSize;
 
+    /** @var bool */
     private $closed = false;
+
+    /** @var bool */
     private $listening = false;
 
     /**
@@ -121,10 +124,10 @@ public function close(): void
     }
 
     /** @internal */
-    public function handleData()
+    public function handleData(): void
     {
         $error = null;
-        \set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error): bool {
+        \set_error_handler(function (int $errno, string $errstr, string $errfile, int $errline) use (&$error): bool {
             $error = new \ErrorException(
                 $errstr,
                 0,

src/ReadableStreamInterface.php

@@ -17,7 +17,7 @@
  *     The event receives a single mixed argument for incoming data.
  *
  *     ```php
- *     $stream->on('data', function ($data) {
+ *     $stream->on('data', function (mixed $data): void {
  *         echo $data;
  *     });
  *     ```
@@ -47,7 +47,7 @@
  *     reached the end of the stream (EOF).
  *
  *     ```php
- *     $stream->on('end', function () {
+ *     $stream->on('end', function (): void {
  *         echo 'END';
  *     });
  *     ```
@@ -84,7 +84,7 @@
  *     The event receives a single `Exception` argument for the error instance.
  *
  *     ```php
- *     $stream->on('error', function (Exception $e) {
+ *     $stream->on('error', function (Exception $e): void {
  *         echo 'Error: ' . $e->getMessage() . PHP_EOL;
  *     });
  *     ```
@@ -116,7 +116,7 @@
  *     The `close` event will be emitted once the stream closes (terminates).
  *
  *     ```php
- *     $stream->on('close', function () {
+ *     $stream->on('close', function (): void {
  *         echo 'CLOSED';
  *     });
  *     ```
@@ -236,7 +236,7 @@ public function pause(): void;
      * ```php
      * $stream->pause();
      *
-     * Loop::addTimer(1.0, function () use ($stream) {
+     * Loop::addTimer(1.0, function () use ($stream): void {
      *     $stream->resume();
      * });
      * ```
@@ -287,7 +287,7 @@ public function resume(): void;
      *
      * ```php
      * $source->pipe($dest);
-     * $source->on('close', function () use ($dest) {
+     * $source->on('close', function () use ($dest): void {
      *     $dest->end('BYE!');
      * });
      * ```
@@ -319,7 +319,7 @@ public function resume(): void;
      * a `pipe` event with this source stream an event argument.
      *
      * @param WritableStreamInterface $dest
-     * @param array $options
+     * @param array{end?:bool} $options
      * @return WritableStreamInterface $dest stream as-is
      */
     public function pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface;