From 74dd76353dde1eec7bba9a844273938cd099ecf7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Christian=20L=C3=BCck?= <christian@clue.engineering>
Date: Wed, 15 May 2024 20:18:59 +0200
Subject: [PATCH] Add native types to public API

---
 README.md                         |  2 +-
 src/CompositeStream.php           | 16 ++++++++--------
 src/DuplexResourceStream.php      | 20 ++++++++++----------
 src/ReadableResourceStream.php    | 14 +++++++-------
 src/ReadableStreamInterface.php   | 10 +++++-----
 src/ThroughStream.php             | 22 +++++++++-------------
 src/Util.php                      | 10 ++++++++--
 src/WritableResourceStream.php    | 14 +++++++-------
 src/WritableStreamInterface.php   |  8 ++++----
 tests/Stub/ReadableStreamStub.php | 10 +++++-----
 tests/ThroughStreamTest.php       |  9 ---------
 11 files changed, 64 insertions(+), 71 deletions(-)

diff --git a/README.md b/README.md
index cce1408..fee7f67 100644
--- a/README.md
+++ b/README.md
@@ -324,7 +324,7 @@ See also `pause()`.
 
 #### pipe()
 
-The `pipe(WritableStreamInterface $dest, array $options = [])` method can be used to
+The `pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface` method can be used to
 pipe all the data from this readable source into the given writable destination.
 
 Automatically sends all incoming data to the destination.
diff --git a/src/CompositeStream.php b/src/CompositeStream.php
index d0b934b..10cef07 100644
--- a/src/CompositeStream.php
+++ b/src/CompositeStream.php
@@ -27,17 +27,17 @@ public function __construct(ReadableStreamInterface $readable, WritableStreamInt
         $this->writable->on('close', [$this, 'close']);
     }
 
-    public function isReadable()
+    public function isReadable(): bool
     {
         return $this->readable->isReadable();
     }
 
-    public function pause()
+    public function pause(): void
     {
         $this->readable->pause();
     }
 
-    public function resume()
+    public function resume(): void
     {
         if (!$this->writable->isWritable()) {
             return;
@@ -46,28 +46,28 @@ public function resume()
         $this->readable->resume();
     }
 
-    public function pipe(WritableStreamInterface $dest, array $options = [])
+    public function pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface
     {
         return Util::pipe($this, $dest, $options);
     }
 
-    public function isWritable()
+    public function isWritable(): bool
     {
         return $this->writable->isWritable();
     }
 
-    public function write($data)
+    public function write($data): bool
     {
         return $this->writable->write($data);
     }
 
-    public function end($data = null)
+    public function end($data = null): void
     {
         $this->readable->pause();
         $this->writable->end($data);
     }
 
-    public function close()
+    public function close(): void
     {
         if ($this->closed) {
             return;
diff --git a/src/DuplexResourceStream.php b/src/DuplexResourceStream.php
index 32a668d..bd05013 100644
--- a/src/DuplexResourceStream.php
+++ b/src/DuplexResourceStream.php
@@ -44,7 +44,7 @@ final class DuplexResourceStream extends EventEmitter implements DuplexStreamInt
      * @param ?int $readChunkSize
      * @param ?WritableStreamInterface $buffer
      */
-    public function __construct($stream, ?LoopInterface $loop = null, $readChunkSize = null, ?WritableStreamInterface $buffer = null)
+    public function __construct($stream, ?LoopInterface $loop = null, ?int $readChunkSize = null, ?WritableStreamInterface $buffer = null)
     {
         if (!\is_resource($stream) || \get_resource_type($stream) !== "stream") {
              throw new InvalidArgumentException('First parameter must be a valid stream resource');
@@ -75,7 +75,7 @@ public function __construct($stream, ?LoopInterface $loop = null, $readChunkSize
 
         $this->stream = $stream;
         $this->loop = $loop ?: Loop::get();
-        $this->bufferSize = ($readChunkSize === null) ? 65536 : (int)$readChunkSize;
+        $this->bufferSize = $readChunkSize ?? 65536;
         $this->buffer = $buffer;
 
         $this->buffer->on('error', function ($error) {
@@ -91,17 +91,17 @@ public function __construct($stream, ?LoopInterface $loop = null, $readChunkSize
         $this->resume();
     }
 
-    public function isReadable()
+    public function isReadable(): bool
     {
         return $this->readable;
     }
 
-    public function isWritable()
+    public function isWritable(): bool
     {
         return $this->writable;
     }
 
-    public function pause()
+    public function pause(): void
     {
         if ($this->listening) {
             $this->loop->removeReadStream($this->stream);
@@ -109,7 +109,7 @@ public function pause()
         }
     }
 
-    public function resume()
+    public function resume(): void
     {
         if (!$this->listening && $this->readable) {
             $this->loop->addReadStream($this->stream, [$this, 'handleData']);
@@ -117,7 +117,7 @@ public function resume()
         }
     }
 
-    public function write($data)
+    public function write($data): bool
     {
         if (!$this->writable) {
             return false;
@@ -126,7 +126,7 @@ public function write($data)
         return $this->buffer->write($data);
     }
 
-    public function close()
+    public function close(): void
     {
         if (!$this->writable && !$this->closing) {
             return;
@@ -147,7 +147,7 @@ public function close()
         }
     }
 
-    public function end($data = null)
+    public function end($data = null): void
     {
         if (!$this->writable) {
             return;
@@ -162,7 +162,7 @@ public function end($data = null)
         $this->buffer->end($data);
     }
 
-    public function pipe(WritableStreamInterface $dest, array $options = [])
+    public function pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface
     {
         return Util::pipe($this, $dest, $options);
     }
diff --git a/src/ReadableResourceStream.php b/src/ReadableResourceStream.php
index 627fbf9..592e35c 100644
--- a/src/ReadableResourceStream.php
+++ b/src/ReadableResourceStream.php
@@ -45,7 +45,7 @@ final class ReadableResourceStream extends EventEmitter implements ReadableStrea
      * @param ?LoopInterface $loop
      * @param ?int $readChunkSize
      */
-    public function __construct($stream, ?LoopInterface $loop = null, $readChunkSize = null)
+    public function __construct($stream, ?LoopInterface $loop = null, ?int $readChunkSize = null)
     {
         if (!\is_resource($stream) || \get_resource_type($stream) !== "stream") {
              throw new InvalidArgumentException('First parameter must be a valid stream resource');
@@ -72,17 +72,17 @@ public function __construct($stream, ?LoopInterface $loop = null, $readChunkSize
 
         $this->stream = $stream;
         $this->loop = $loop ?: Loop::get();
-        $this->bufferSize = ($readChunkSize === null) ? 65536 : (int)$readChunkSize;
+        $this->bufferSize = $readChunkSize ?? 65536;
 
         $this->resume();
     }
 
-    public function isReadable()
+    public function isReadable(): bool
     {
         return !$this->closed;
     }
 
-    public function pause()
+    public function pause(): void
     {
         if ($this->listening) {
             $this->loop->removeReadStream($this->stream);
@@ -90,7 +90,7 @@ public function pause()
         }
     }
 
-    public function resume()
+    public function resume(): void
     {
         if (!$this->listening && !$this->closed) {
             $this->loop->addReadStream($this->stream, [$this, 'handleData']);
@@ -98,12 +98,12 @@ public function resume()
         }
     }
 
-    public function pipe(WritableStreamInterface $dest, array $options = [])
+    public function pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface
     {
         return Util::pipe($this, $dest, $options);
     }
 
-    public function close()
+    public function close(): void
     {
         if ($this->closed) {
             return;
diff --git a/src/ReadableStreamInterface.php b/src/ReadableStreamInterface.php
index ecc5267..e4ddf32 100644
--- a/src/ReadableStreamInterface.php
+++ b/src/ReadableStreamInterface.php
@@ -194,7 +194,7 @@ interface ReadableStreamInterface extends EventEmitterInterface
      *
      * @return bool
      */
-    public function isReadable();
+    public function isReadable(): bool;
 
     /**
      * Pauses reading incoming data events.
@@ -226,7 +226,7 @@ public function isReadable();
      * @see self::resume()
      * @return void
      */
-    public function pause();
+    public function pause(): void;
 
     /**
      * Resumes reading incoming data events.
@@ -247,7 +247,7 @@ public function pause();
      * @see self::pause()
      * @return void
      */
-    public function resume();
+    public function resume(): void;
 
     /**
      * Pipes all the data from this readable source into the given writable destination.
@@ -322,7 +322,7 @@ public function resume();
      * @param array $options
      * @return WritableStreamInterface $dest stream as-is
      */
-    public function pipe(WritableStreamInterface $dest, array $options = []);
+    public function pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface;
 
     /**
      * Closes the stream (forcefully).
@@ -358,5 +358,5 @@ public function pipe(WritableStreamInterface $dest, array $options = []);
      * @return void
      * @see WritableStreamInterface::close()
      */
-    public function close();
+    public function close(): void;
 }
diff --git a/src/ThroughStream.php b/src/ThroughStream.php
index c49ebfc..4c1f4a2 100644
--- a/src/ThroughStream.php
+++ b/src/ThroughStream.php
@@ -82,22 +82,18 @@ final class ThroughStream extends EventEmitter implements DuplexStreamInterface
     private $drain = false;
     private $callback;
 
-    public function __construct($callback = null)
+    public function __construct(?callable $callback = null)
     {
-        if ($callback !== null && !\is_callable($callback)) {
-            throw new InvalidArgumentException('Invalid transformation callback given');
-        }
-
         $this->callback = $callback;
     }
 
-    public function pause()
+    public function pause(): void
     {
         // only allow pause if still readable, false otherwise
         $this->paused = $this->readable;
     }
 
-    public function resume()
+    public function resume(): void
     {
         $this->paused = false;
 
@@ -108,22 +104,22 @@ public function resume()
         }
     }
 
-    public function pipe(WritableStreamInterface $dest, array $options = [])
+    public function pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface
     {
         return Util::pipe($this, $dest, $options);
     }
 
-    public function isReadable()
+    public function isReadable(): bool
     {
         return $this->readable;
     }
 
-    public function isWritable()
+    public function isWritable(): bool
     {
         return $this->writable;
     }
 
-    public function write($data)
+    public function write($data): bool
     {
         if (!$this->writable) {
             return false;
@@ -151,7 +147,7 @@ public function write($data)
         return $this->writable && !$this->paused;
     }
 
-    public function end($data = null)
+    public function end($data = null): void
     {
         if (!$this->writable) {
             return;
@@ -175,7 +171,7 @@ public function end($data = null)
         $this->close();
     }
 
-    public function close()
+    public function close(): void
     {
         if ($this->closed) {
             return;
diff --git a/src/Util.php b/src/Util.php
index 114ccb8..23dd252 100644
--- a/src/Util.php
+++ b/src/Util.php
@@ -13,7 +13,7 @@ final class Util
      * @return WritableStreamInterface $dest stream as-is
      * @see ReadableStreamInterface::pipe() for more details
      */
-    public static function pipe(ReadableStreamInterface $source, WritableStreamInterface $dest, array $options = [])
+    public static function pipe(ReadableStreamInterface $source, WritableStreamInterface $dest, array $options = []): WritableStreamInterface
     {
         // source not readable => NO-OP
         if (!$source->isReadable()) {
@@ -64,7 +64,13 @@ public static function pipe(ReadableStreamInterface $source, WritableStreamInter
         return $dest;
     }
 
-    public static function forwardEvents($source, $target, array $events)
+    /**
+     * @param ReadableStreamInterface|WritableStreamInterface $source
+     * @param ReadableStreamInterface|WritableStreamInterface $target
+     * @param string[] $events
+     * @return void
+     */
+    public static function forwardEvents($source, $target, array $events): void
     {
         foreach ($events as $event) {
             $source->on($event, function () use ($event, $target) {
diff --git a/src/WritableResourceStream.php b/src/WritableResourceStream.php
index 9792a5f..84ec4f6 100644
--- a/src/WritableResourceStream.php
+++ b/src/WritableResourceStream.php
@@ -34,7 +34,7 @@ final class WritableResourceStream extends EventEmitter implements WritableStrea
      * @param ?int $writeBufferSoftLimit
      * @param ?int $writeChunkSize
      */
-    public function __construct($stream, ?LoopInterface $loop = null, $writeBufferSoftLimit = null, $writeChunkSize = null)
+    public function __construct($stream, ?LoopInterface $loop = null, ?int $writeBufferSoftLimit = null, ?int $writeChunkSize = null)
     {
         if (!\is_resource($stream) || \get_resource_type($stream) !== "stream") {
             throw new \InvalidArgumentException('First parameter must be a valid stream resource');
@@ -54,16 +54,16 @@ public function __construct($stream, ?LoopInterface $loop = null, $writeBufferSo
 
         $this->stream = $stream;
         $this->loop = $loop ?: Loop::get();
-        $this->softLimit = ($writeBufferSoftLimit === null) ? 65536 : (int)$writeBufferSoftLimit;
-        $this->writeChunkSize = ($writeChunkSize === null) ? -1 : (int)$writeChunkSize;
+        $this->softLimit = $writeBufferSoftLimit ?? 65536;
+        $this->writeChunkSize = $writeChunkSize ?? -1;
     }
 
-    public function isWritable()
+    public function isWritable(): bool
     {
         return $this->writable;
     }
 
-    public function write($data)
+    public function write($data): bool
     {
         if (!$this->writable) {
             return false;
@@ -80,7 +80,7 @@ public function write($data)
         return !isset($this->data[$this->softLimit - 1]);
     }
 
-    public function end($data = null)
+    public function end($data = null): void
     {
         if (null !== $data) {
             $this->write($data);
@@ -95,7 +95,7 @@ public function end($data = null)
         }
     }
 
-    public function close()
+    public function close(): void
     {
         if ($this->closed) {
             return;
diff --git a/src/WritableStreamInterface.php b/src/WritableStreamInterface.php
index e262592..77baaad 100644
--- a/src/WritableStreamInterface.php
+++ b/src/WritableStreamInterface.php
@@ -170,7 +170,7 @@ interface WritableStreamInterface extends EventEmitterInterface
      *
      * @return bool
      */
-    public function isWritable();
+    public function isWritable(): bool;
 
     /**
      * Write some data into the stream.
@@ -219,7 +219,7 @@ public function isWritable();
      * @param mixed|string $data
      * @return bool
      */
-    public function write($data);
+    public function write($data): bool;
 
     /**
      * Successfully ends the stream (after optionally sending some final data).
@@ -292,7 +292,7 @@ public function write($data);
      * @param mixed|string|null $data
      * @return void
      */
-    public function end($data = null);
+    public function end($data = null): void;
 
     /**
      * Closes the stream (forcefully).
@@ -343,5 +343,5 @@ public function end($data = null);
      * @return void
      * @see ReadableStreamInterface::close()
      */
-    public function close();
+    public function close(): void;
 }
diff --git a/tests/Stub/ReadableStreamStub.php b/tests/Stub/ReadableStreamStub.php
index 3fa56a8..669059c 100644
--- a/tests/Stub/ReadableStreamStub.php
+++ b/tests/Stub/ReadableStreamStub.php
@@ -12,7 +12,7 @@ class ReadableStreamStub extends EventEmitter implements ReadableStreamInterface
     public $readable = true;
     public $paused = false;
 
-    public function isReadable()
+    public function isReadable(): bool
     {
         return true;
     }
@@ -35,24 +35,24 @@ public function end()
         $this->emit('end', []);
     }
 
-    public function pause()
+    public function pause(): void
     {
         $this->paused = true;
     }
 
-    public function resume()
+    public function resume(): void
     {
         $this->paused = false;
     }
 
-    public function close()
+    public function close(): void
     {
         $this->readable = false;
 
         $this->emit('close');
     }
 
-    public function pipe(WritableStreamInterface $dest, array $options = [])
+    public function pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface
     {
         Util::pipe($this, $dest, $options);
 
diff --git a/tests/ThroughStreamTest.php b/tests/ThroughStreamTest.php
index 04a49d0..85c22de 100644
--- a/tests/ThroughStreamTest.php
+++ b/tests/ThroughStreamTest.php
@@ -10,15 +10,6 @@
  */
 class ThroughStreamTest extends TestCase
 {
-    /**
-     * @test
-     */
-    public function itShouldRejectInvalidCallback()
-    {
-        $this->expectException(\InvalidArgumentException::class);
-        new ThroughStream(123);
-    }
-
     /** @test */
     public function itShouldReturnTrueForAnyDataWrittenToIt()
     {