From 9234da07cb99728b5530c0534176d0dc4c847d5b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Christian=20L=C3=BCck?= <christian@clue.engineering>
Date: Wed, 15 May 2024 20:35:55 +0200
Subject: [PATCH 1/2] Add PHPStan to test environment

---
 .gitattributes                                |  1 +
 .github/workflows/ci.yml                      | 23 +++++++++++++++++++
 README.md                                     |  6 +++++
 composer.json                                 |  5 ++--
 phpstan.neon.dist                             |  7 ++++++
 src/DuplexResourceStream.php                  |  3 ++-
 src/ReadableResourceStream.php                |  3 ++-
 src/ThroughStream.php                         |  2 ++
 src/WritableResourceStream.php                |  3 ++-
 tests/CompositeStreamTest.php                 | 18 +++++++++++++++
 tests/DuplexResourceStreamIntegrationTest.php | 22 +++++++++---------
 tests/DuplexResourceStreamTest.php            |  7 +++++-
 tests/ReadableResourceStreamTest.php          | 17 +++++++-------
 tests/ThroughStreamTest.php                   |  3 ++-
 tests/UtilTest.php                            | 16 +++++++++++++
 tests/WritableResourceStreamTest.php          |  3 ++-
 16 files changed, 112 insertions(+), 27 deletions(-)
 create mode 100644 phpstan.neon.dist

diff --git a/.gitattributes b/.gitattributes
index fc0be87..e3b263e 100644
--- a/.gitattributes
+++ b/.gitattributes
@@ -2,6 +2,7 @@
 /.github/ export-ignore
 /.gitignore export-ignore
 /examples/ export-ignore
+/phpstan.neon.dist export-ignore
 /phpunit.xml.dist export-ignore
 /phpunit.xml.legacy export-ignore
 /tests/ export-ignore
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f654fb0..1bb8701 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -47,3 +47,26 @@ jobs:
       - run: composer install
       - run: vendor/bin/phpunit --coverage-text
       - run: time php examples/91-benchmark-throughput.php
+
+  PHPStan:
+    name: PHPStan (PHP ${{ matrix.php }})
+    runs-on: ubuntu-22.04
+    strategy:
+      matrix:
+        php:
+          - 8.3
+          - 8.2
+          - 8.1
+          - 8.0
+          - 7.4
+          - 7.3
+          - 7.2
+          - 7.1
+    steps:
+      - uses: actions/checkout@v4
+      - uses: shivammathur/setup-php@v2
+        with:
+          php-version: ${{ matrix.php }}
+          coverage: none
+      - run: composer install
+      - run: vendor/bin/phpstan
diff --git a/README.md b/README.md
index fee7f67..e27cbec 100644
--- a/README.md
+++ b/README.md
@@ -1243,6 +1243,12 @@ 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:
+
+```bash
+vendor/bin/phpstan
+```
+
 ## License
 
 MIT, see [LICENSE file](LICENSE).
diff --git a/composer.json b/composer.json
index b8d34d4..beb0a56 100644
--- a/composer.json
+++ b/composer.json
@@ -31,8 +31,9 @@
         "evenement/evenement": "^3.0 || ^2.0 || ^1.0"
     },
     "require-dev": {
-        "phpunit/phpunit": "^9.6 || ^7.5",
-        "clue/stream-filter": "~1.2"
+        "clue/stream-filter": "^1.2",
+        "phpstan/phpstan": "1.11.1 || 1.4.10",
+        "phpunit/phpunit": "^9.6 || ^7.5"
     },
     "autoload": {
         "psr-4": {
diff --git a/phpstan.neon.dist b/phpstan.neon.dist
new file mode 100644
index 0000000..d631a6d
--- /dev/null
+++ b/phpstan.neon.dist
@@ -0,0 +1,7 @@
+parameters:
+    level: 5
+
+    paths:
+        - examples/
+        - src/
+        - tests/
diff --git a/src/DuplexResourceStream.php b/src/DuplexResourceStream.php
index bd05013..94d013c 100644
--- a/src/DuplexResourceStream.php
+++ b/src/DuplexResourceStream.php
@@ -171,7 +171,7 @@ public function pipe(WritableStreamInterface $dest, array $options = []): Writab
     public function handleData($stream)
     {
         $error = null;
-        \set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error) {
+        \set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error): bool {
             $error = new \ErrorException(
                 $errstr,
                 0,
@@ -179,6 +179,7 @@ public function handleData($stream)
                 $errfile,
                 $errline
             );
+            return true;
         });
 
         $data = \stream_get_contents($stream, $this->bufferSize);
diff --git a/src/ReadableResourceStream.php b/src/ReadableResourceStream.php
index 592e35c..a0993bb 100644
--- a/src/ReadableResourceStream.php
+++ b/src/ReadableResourceStream.php
@@ -124,7 +124,7 @@ public function close(): void
     public function handleData()
     {
         $error = null;
-        \set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error) {
+        \set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error): bool {
             $error = new \ErrorException(
                 $errstr,
                 0,
@@ -132,6 +132,7 @@ public function handleData()
                 $errfile,
                 $errline
             );
+            return true;
         });
 
         $data = \stream_get_contents($this->stream, $this->bufferSize);
diff --git a/src/ThroughStream.php b/src/ThroughStream.php
index 4c1f4a2..a209982 100644
--- a/src/ThroughStream.php
+++ b/src/ThroughStream.php
@@ -144,6 +144,7 @@ public function write($data): bool
         }
 
         // continue writing if still writable and not paused (throttled), false otherwise
+        // @phpstan-ignore-next-line (may be false when write() causes stream to close)
         return $this->writable && !$this->paused;
     }
 
@@ -157,6 +158,7 @@ public function end($data = null): void
             $this->write($data);
 
             // return if write() already caused the stream to close
+            // @phpstan-ignore-next-line (may be false when write() causes stream to close)
             if (!$this->writable) {
                 return;
             }
diff --git a/src/WritableResourceStream.php b/src/WritableResourceStream.php
index 84ec4f6..6140246 100644
--- a/src/WritableResourceStream.php
+++ b/src/WritableResourceStream.php
@@ -122,8 +122,9 @@ public function close(): void
     public function handleWrite()
     {
         $error = null;
-        \set_error_handler(function ($_, $errstr) use (&$error) {
+        \set_error_handler(function ($_, $errstr) use (&$error): bool {
             $error = $errstr;
+            return true;
         });
 
         if ($this->writeChunkSize === -1) {
diff --git a/tests/CompositeStreamTest.php b/tests/CompositeStreamTest.php
index 75a0426..cc9ef28 100644
--- a/tests/CompositeStreamTest.php
+++ b/tests/CompositeStreamTest.php
@@ -23,12 +23,14 @@ public function itShouldCloseReadableIfNotWritable()
         $readable
             ->expects($this->once())
             ->method('close');
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
             ->expects($this->once())
             ->method('isWritable')
             ->willReturn(false);
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
 
@@ -44,11 +46,13 @@ public function itShouldCloseWritableIfNotReadable()
             ->expects($this->once())
             ->method('isReadable')
             ->willReturn(false);
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
             ->expects($this->once())
             ->method('close');
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
 
@@ -64,6 +68,7 @@ public function itShouldForwardWritableCallsToWritableStream()
             ->expects($this->once())
             ->method('isReadable')
             ->willReturn(true);
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
@@ -74,6 +79,7 @@ public function itShouldForwardWritableCallsToWritableStream()
             ->expects($this->exactly(2))
             ->method('isWritable')
             ->willReturn(true);
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
         $composite->write('foo');
@@ -94,12 +100,14 @@ public function itShouldForwardReadableCallsToReadableStream()
         $readable
             ->expects($this->once())
             ->method('resume');
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
             ->expects($this->any())
             ->method('isWritable')
             ->willReturn(true);
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
         $composite->isReadable();
@@ -118,12 +126,14 @@ public function itShouldNotForwardResumeIfStreamIsNotWritable()
         $readable
             ->expects($this->never())
             ->method('resume');
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
             ->expects($this->exactly(2))
             ->method('isWritable')
             ->willReturnOnConsecutiveCalls(true, false);
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
         $composite->resume();
@@ -137,6 +147,7 @@ public function endShouldDelegateToWritableWithData()
             ->expects($this->once())
             ->method('isReadable')
             ->willReturn(true);
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
@@ -147,6 +158,7 @@ public function endShouldDelegateToWritableWithData()
             ->expects($this->once())
             ->method('end')
             ->with('foo');
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
         $composite->end('foo');
@@ -163,6 +175,7 @@ public function closeShouldCloseBothStreams()
         $readable
             ->expects($this->once())
             ->method('close');
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
@@ -172,6 +185,7 @@ public function closeShouldCloseBothStreams()
         $writable
             ->expects($this->once())
             ->method('close');
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
         $composite->close();
@@ -231,6 +245,7 @@ public function itShouldHandlePipingCorrectly()
             ->expects($this->once())
             ->method('isReadable')
             ->willReturn(true);
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable->expects($this->any())->method('isWritable')->willReturn(True);
@@ -238,6 +253,7 @@ public function itShouldHandlePipingCorrectly()
             ->expects($this->once())
             ->method('write')
             ->with('foo');
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
 
@@ -253,6 +269,7 @@ public function itShouldForwardPipeCallsToReadableStream()
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable->expects($this->any())->method('isWritable')->willReturn(True);
+        assert($writable instanceof WritableStreamInterface);
 
         $composite = new CompositeStream($readable, $writable);
 
@@ -262,6 +279,7 @@ public function itShouldForwardPipeCallsToReadableStream()
             ->expects($this->once())
             ->method('write')
             ->with('foo');
+        assert($output instanceof WritableStreamInterface);
 
         $composite->pipe($output);
         $readable->emit('data', ['foo']);
diff --git a/tests/DuplexResourceStreamIntegrationTest.php b/tests/DuplexResourceStreamIntegrationTest.php
index 02da5d2..a468aef 100644
--- a/tests/DuplexResourceStreamIntegrationTest.php
+++ b/tests/DuplexResourceStreamIntegrationTest.php
@@ -37,7 +37,7 @@ function () {
     public function testBufferReadsLargeChunks($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -73,7 +73,7 @@ public function testBufferReadsLargeChunks($condition, $loopFactory)
     public function testWriteLargeChunk($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -113,7 +113,7 @@ public function testWriteLargeChunk($condition, $loopFactory)
     public function testDoesNotEmitDataIfNothingHasBeenWritten($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -141,7 +141,7 @@ public function testDoesNotEmitDataIfNothingHasBeenWritten($condition, $loopFact
     public function testDoesNotWriteDataIfRemoteSideFromPairHasBeenClosed($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -171,7 +171,7 @@ public function testDoesNotWriteDataIfRemoteSideFromPairHasBeenClosed($condition
     public function testDoesNotWriteDataIfServerSideHasBeenClosed($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -204,7 +204,7 @@ public function testDoesNotWriteDataIfServerSideHasBeenClosed($condition, $loopF
     public function testDoesNotWriteDataIfClientSideHasBeenClosed($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -237,7 +237,7 @@ public function testDoesNotWriteDataIfClientSideHasBeenClosed($condition, $loopF
     public function testReadsSingleChunkFromProcessPipe($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -256,7 +256,7 @@ public function testReadsSingleChunkFromProcessPipe($condition, $loopFactory)
     public function testReadsMultipleChunksFromProcessPipe($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -282,7 +282,7 @@ public function testReadsMultipleChunksFromProcessPipe($condition, $loopFactory)
     public function testReadsLongChunksFromProcessPipe($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -308,7 +308,7 @@ public function testReadsLongChunksFromProcessPipe($condition, $loopFactory)
     public function testReadsNothingFromProcessPipeWithNoOutput($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $loop = $loopFactory();
@@ -328,7 +328,7 @@ public function testReadsNothingFromProcessPipeWithNoOutput($condition, $loopFac
     public function testEmptyReadShouldntFcloseStream($condition, $loopFactory)
     {
         if (true !== $condition()) {
-            return $this->markTestSkipped('Loop implementation not available');
+            $this->markTestSkipped('Loop implementation not available');
         }
 
         $server = stream_socket_server('tcp://127.0.0.1:0');
diff --git a/tests/DuplexResourceStreamTest.php b/tests/DuplexResourceStreamTest.php
index 52477fe..5b18553 100644
--- a/tests/DuplexResourceStreamTest.php
+++ b/tests/DuplexResourceStreamTest.php
@@ -59,7 +59,7 @@ public function testConstructorThrowsExceptionOnInvalidStream()
         $loop = $this->createLoopMock();
 
         $this->expectException(\InvalidArgumentException::class);
-        new DuplexResourceStream('breakme', $loop);
+        new DuplexResourceStream('breakme', $loop); // @phpstan-ignore-line
     }
 
     /**
@@ -114,6 +114,7 @@ public function testConstructorAcceptsBuffer()
         $loop = $this->createLoopMock();
 
         $buffer = $this->createMock(WritableStreamInterface::class);
+        assert($buffer instanceof WritableStreamInterface);
 
         new DuplexResourceStream($stream, $loop, null, $buffer);
     }
@@ -131,6 +132,7 @@ public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlockingW
         $loop = $this->createLoopMock();
 
         $buffer = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        assert($buffer instanceof WritableStreamInterface);
 
         $this->expectException(\RuntimeException::class);
         new DuplexResourceStream($stream, $loop, null, $buffer);
@@ -157,6 +159,7 @@ public function testEndShouldEndBuffer()
 
         $buffer = $this->createMock(WritableStreamInterface::class);
         $buffer->expects($this->once())->method('end')->with('foo');
+        assert($buffer instanceof WritableStreamInterface);
 
         $conn = new DuplexResourceStream($stream, $loop, null, $buffer);
         $conn->end('foo');
@@ -170,6 +173,7 @@ public function testEndAfterCloseIsNoOp()
 
         $buffer = $this->createMock(WritableStreamInterface::class);
         $buffer->expects($this->never())->method('end');
+        assert($buffer instanceof WritableStreamInterface);
 
         $conn = new DuplexResourceStream($stream, $loop);
         $conn->close();
@@ -409,6 +413,7 @@ public function testPipeShouldReturnDestination()
 
         $conn = new DuplexResourceStream($stream, $loop);
         $dest = $this->createMock(WritableStreamInterface::class);
+        assert($dest instanceof WritableStreamInterface);
 
         $this->assertSame($dest, $conn->pipe($dest));
     }
diff --git a/tests/ReadableResourceStreamTest.php b/tests/ReadableResourceStreamTest.php
index 4cebb5a..365ac0d 100644
--- a/tests/ReadableResourceStreamTest.php
+++ b/tests/ReadableResourceStreamTest.php
@@ -58,7 +58,7 @@ public function testConstructorThrowsExceptionOnInvalidStream()
         $loop = $this->createLoopMock();
 
         $this->expectException(\InvalidArgumentException::class);
-        new ReadableResourceStream(false, $loop);
+        new ReadableResourceStream(false, $loop); // @phpstan-ignore-line
     }
 
     /**
@@ -148,7 +148,7 @@ public function testDataEvent()
         fwrite($stream, "foobar\n");
         rewind($stream);
 
-        $conn->handleData($stream);
+        $conn->handleData();
         $this->assertSame("foobar\n", $capturedData);
     }
 
@@ -171,7 +171,7 @@ public function testDataEventDoesEmitOneChunkMatchingBufferSize()
         fwrite($stream, str_repeat("a", 100000));
         rewind($stream);
 
-        $conn->handleData($stream);
+        $conn->handleData();
 
         $this->assertTrue($conn->isReadable());
         $this->assertEquals(4321, strlen($capturedData));
@@ -197,7 +197,7 @@ public function testDataEventDoesEmitOneChunkUntilStreamEndsWhenBufferSizeIsInfi
         fwrite($stream, str_repeat("a", 100000));
         rewind($stream);
 
-        $conn->handleData($stream);
+        $conn->handleData();
 
         $this->assertTrue($conn->isReadable());
         $this->assertEquals(100000, strlen($capturedData));
@@ -214,7 +214,7 @@ public function testEmptyStreamShouldNotEmitData()
         $conn = new ReadableResourceStream($stream, $loop);
         $conn->on('data', $this->expectCallableNever());
 
-        $conn->handleData($stream);
+        $conn->handleData();
     }
 
     public function testPipeShouldReturnDestination()
@@ -224,6 +224,7 @@ public function testPipeShouldReturnDestination()
 
         $conn = new ReadableResourceStream($stream, $loop);
         $dest = $this->createMock(WritableStreamInterface::class);
+        assert($dest instanceof WritableStreamInterface);
 
         $this->assertSame($dest, $conn->pipe($dest));
     }
@@ -245,7 +246,7 @@ public function testClosingStreamInDataEventShouldNotTriggerError()
         fwrite($stream, "foobar\n");
         rewind($stream);
 
-        $conn->handleData($stream);
+        $conn->handleData();
     }
 
     /**
@@ -344,7 +345,7 @@ public function testDataFiltered()
         fwrite($stream, "foobar\n");
         rewind($stream);
 
-        $conn->handleData($stream);
+        $conn->handleData();
         $this->assertSame("foobr\n", $capturedData);
     }
 
@@ -373,7 +374,7 @@ public function testDataErrorShouldEmitErrorAndClose()
         fwrite($stream, "foobar\n");
         rewind($stream);
 
-        $conn->handleData($stream);
+        $conn->handleData();
     }
 
     /**
diff --git a/tests/ThroughStreamTest.php b/tests/ThroughStreamTest.php
index 85c22de..ddd570f 100644
--- a/tests/ThroughStreamTest.php
+++ b/tests/ThroughStreamTest.php
@@ -216,7 +216,7 @@ public function endShouldWriteDataBeforeClosing()
     public function endTwiceShouldOnlyEmitOnce()
     {
         $through = new ThroughStream();
-        $through->on('data', $this->expectCallableOnce('first'));
+        $through->on('data', $this->expectCallableOnceWith('first'));
         $through->end('first');
         $through->end('ignored');
     }
@@ -309,6 +309,7 @@ public function pipeShouldPipeCorrectly()
             ->expects($this->once())
             ->method('write')
             ->with('foo');
+        assert($output instanceof WritableStreamInterface);
 
         $through = new ThroughStream();
         $through->pipe($output);
diff --git a/tests/UtilTest.php b/tests/UtilTest.php
index 50b2e75..cd00c8b 100644
--- a/tests/UtilTest.php
+++ b/tests/UtilTest.php
@@ -18,8 +18,10 @@ class UtilTest extends TestCase
     public function testPipeReturnsDestinationStream()
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
+        assert($writable instanceof WritableStreamInterface);
 
         $ret = Util::pipe($readable, $writable);
 
@@ -33,6 +35,7 @@ public function testPipeNonReadableSourceShouldDoNothing()
             ->expects($this->any())
             ->method('isReadable')
             ->willReturn(false);
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
@@ -41,6 +44,7 @@ public function testPipeNonReadableSourceShouldDoNothing()
         $writable
             ->expects($this->never())
             ->method('end');
+        assert($writable instanceof WritableStreamInterface);
 
         Util::pipe($readable, $writable);
     }
@@ -55,6 +59,7 @@ public function testPipeIntoNonWritableDestinationShouldPauseSource()
         $readable
             ->expects($this->once())
             ->method('pause');
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable
@@ -64,6 +69,7 @@ public function testPipeIntoNonWritableDestinationShouldPauseSource()
         $writable
             ->expects($this->never())
             ->method('end');
+        assert($writable instanceof WritableStreamInterface);
 
         Util::pipe($readable, $writable);
     }
@@ -78,6 +84,7 @@ public function testPipeClosingDestPausesSource()
         $readable
             ->expects($this->once())
             ->method('pause');
+        assert($readable instanceof ReadableStreamInterface);
 
         $writable = new ThroughStream();
 
@@ -98,6 +105,7 @@ public function testPipeWithEnd()
         $writable
             ->expects($this->once())
             ->method('end');
+        assert($writable instanceof WritableStreamInterface);
 
         Util::pipe($readable, $writable);
 
@@ -116,6 +124,7 @@ public function testPipeWithoutEnd()
         $writable
             ->expects($this->never())
             ->method('end');
+        assert($writable instanceof WritableStreamInterface);
 
         Util::pipe($readable, $writable, ['end' => false]);
 
@@ -136,6 +145,7 @@ public function testPipeWithTooSlowWritableShouldPauseReadable()
             ->method('write')
             ->with('some data')
             ->will($this->returnValue(false));
+        assert($writable instanceof WritableStreamInterface);
 
         $readable->pipe($writable);
 
@@ -163,6 +173,7 @@ public function testPipeWithTooSlowWritableShouldResumeOnDrain()
                     $onDrain = $callback;
                 }
             }));
+        assert($writable instanceof WritableStreamInterface);
 
         $readable->pipe($writable);
         $readable->pause();
@@ -179,6 +190,7 @@ public function testPipeWithWritableResourceStream()
 
         $stream = fopen('php://temp', 'r+');
         $loop = $this->createMock(LoopInterface::class);
+        assert($loop instanceof LoopInterface);
         $buffer = new WritableResourceStream($stream, $loop);
 
         $readable->pipe($buffer);
@@ -239,8 +251,12 @@ public function testPipeDuplexIntoSelfEndsOnEnd()
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable->expects($this->any())->method('isReadable')->willReturn(true);
+        assert($readable instanceof ReadableStreamInterface);
+
         $writable = $this->createMock(WritableStreamInterface::class);
         $writable->expects($this->any())->method('isWritable')->willReturn(true);
+        assert($writable instanceof WritableStreamInterface);
+
         $duplex = new CompositeStream($readable, $writable);
 
         Util::pipe($duplex, $duplex);
diff --git a/tests/WritableResourceStreamTest.php b/tests/WritableResourceStreamTest.php
index 02f2367..df3e815 100644
--- a/tests/WritableResourceStreamTest.php
+++ b/tests/WritableResourceStreamTest.php
@@ -58,7 +58,7 @@ public function testConstructorThrowsIfNotAValidStreamResource()
         $loop = $this->createLoopMock();
 
         $this->expectException(\InvalidArgumentException::class);
-        new WritableResourceStream($stream, $loop);
+        new WritableResourceStream($stream, $loop); // @phpstan-ignore-line
     }
 
     /**
@@ -165,6 +165,7 @@ public function testWriteReturnsFalseWhenWritableResourceStreamIsFull()
             ->expects($this->any())
             ->method('addWriteStream')
             ->will($this->returnCallback(function ($stream, $listener) use (&$preventWrites) {
+                /** @var bool $preventWrites */
                 if (!$preventWrites) {
                     call_user_func($listener, $stream);
                 }

From 5047943c375e568f52ae67bb3a5abfb60af65025 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Christian=20L=C3=BCck?= <christian@clue.engineering>
Date: Wed, 15 May 2024 21:15:56 +0200
Subject: [PATCH 2/2] Improve type definitions and update to PHPStan level
 `max`

---
 README.md                                     |  36 ++---
 examples/01-http.php                          |   4 +-
 examples/02-https.php                         |   4 +-
 examples/91-benchmark-throughput.php          |  12 +-
 phpstan.neon.dist                             |   2 +-
 src/CompositeStream.php                       |   5 +
 src/DuplexResourceStream.php                  |  23 +++-
 src/ReadableResourceStream.php                |   7 +-
 src/ReadableStreamInterface.php               |  14 +-
 src/ThroughStream.php                         |  19 ++-
 src/Util.php                                  |  16 +--
 src/WritableResourceStream.php                |  23 ++--
 src/WritableStreamInterface.php               |  12 +-
 tests/CompositeStreamTest.php                 |  24 ++--
 tests/DuplexResourceStreamIntegrationTest.php |  80 +++++++----
 tests/DuplexResourceStreamTest.php            | 128 +++++++++++++-----
 tests/EnforceBlockingWrapper.php              |   9 +-
 tests/FunctionalInternetTest.php              |  14 +-
 tests/ReadableResourceStreamTest.php          |  98 ++++++++++----
 tests/Stub/ReadableStreamStub.php             |  16 ++-
 tests/TestCase.php                            |  18 ++-
 tests/ThroughStreamTest.php                   |  56 ++++----
 tests/UtilTest.php                            |  30 ++--
 tests/WritableResourceStreamTest.php          | 124 ++++++++++++-----
 24 files changed, 511 insertions(+), 263 deletions(-)

diff --git a/README.md b/README.md
index e27cbec..870a907 100644
--- a/README.md
+++ b/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
diff --git a/examples/01-http.php b/examples/01-http.php
index 786aefc..316e7e3 100644
--- a/examples/01-http.php
+++ b/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;
 });
 
diff --git a/examples/02-https.php b/examples/02-https.php
index e629791..8b3ad79 100644
--- a/examples/02-https.php
+++ b/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;
 });
 
diff --git a/examples/91-benchmark-throughput.php b/examples/91-benchmark-throughput.php
index c396606..3d0213c 100644
--- a/examples/91-benchmark-throughput.php
+++ b/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);
 
diff --git a/phpstan.neon.dist b/phpstan.neon.dist
index d631a6d..0fe275f 100644
--- a/phpstan.neon.dist
+++ b/phpstan.neon.dist
@@ -1,5 +1,5 @@
 parameters:
-    level: 5
+    level: max
 
     paths:
         - examples/
diff --git a/src/CompositeStream.php b/src/CompositeStream.php
index 10cef07..db6db78 100644
--- a/src/CompositeStream.php
+++ b/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)
diff --git a/src/DuplexResourceStream.php b/src/DuplexResourceStream.php
index 94d013c..5990906 100644
--- a/src/DuplexResourceStream.php
+++ b/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,
diff --git a/src/ReadableResourceStream.php b/src/ReadableResourceStream.php
index a0993bb..cf1161a 100644
--- a/src/ReadableResourceStream.php
+++ b/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,
diff --git a/src/ReadableStreamInterface.php b/src/ReadableStreamInterface.php
index e4ddf32..94e5915 100644
--- a/src/ReadableStreamInterface.php
+++ b/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;
diff --git a/src/ThroughStream.php b/src/ThroughStream.php
index a209982..a8e4837 100644
--- a/src/ThroughStream.php
+++ b/src/ThroughStream.php
@@ -3,7 +3,6 @@
 namespace React\Stream;
 
 use Evenement\EventEmitter;
-use InvalidArgumentException;
 
 /**
  * The `ThroughStream` implements the
@@ -43,7 +42,7 @@
  * 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"));
@@ -55,7 +54,7 @@
  * 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');
  *     }
@@ -75,13 +74,27 @@
  */
 final class ThroughStream extends EventEmitter implements DuplexStreamInterface
 {
+    /** @var bool */
     private $readable = true;
+
+    /** @var bool */
     private $writable = true;
+
+    /** @var bool */
     private $closed = false;
+
+    /** @var bool */
     private $paused = false;
+
+    /** @var bool */
     private $drain = false;
+
+    /** @var ?callable */
     private $callback;
 
+    /**
+     * @param ?callable $callback
+     */
     public function __construct(?callable $callback = null)
     {
         $this->callback = $callback;
diff --git a/src/Util.php b/src/Util.php
index 23dd252..c6c038d 100644
--- a/src/Util.php
+++ b/src/Util.php
@@ -9,7 +9,7 @@ final class Util
      *
      * @param ReadableStreamInterface $source
      * @param WritableStreamInterface $dest
-     * @param array $options
+     * @param array{end?:bool} $options
      * @return WritableStreamInterface $dest stream as-is
      * @see ReadableStreamInterface::pipe() for more details
      */
@@ -30,33 +30,33 @@ public static function pipe(ReadableStreamInterface $source, WritableStreamInter
         $dest->emit('pipe', [$source]);
 
         // forward all source data events as $dest->write()
-        $source->on('data', $dataer = function ($data) use ($source, $dest) {
+        $source->on('data', $dataer = function ($data) use ($source, $dest): void {
             $feedMore = $dest->write($data);
 
             if (false === $feedMore) {
                 $source->pause();
             }
         });
-        $dest->on('close', function () use ($source, $dataer) {
+        $dest->on('close', function () use ($source, $dataer): void {
             $source->removeListener('data', $dataer);
             $source->pause();
         });
 
         // forward destination drain as $source->resume()
-        $dest->on('drain', $drainer = function () use ($source) {
+        $dest->on('drain', $drainer = function () use ($source): void {
             $source->resume();
         });
-        $source->on('close', function () use ($dest, $drainer) {
+        $source->on('close', function () use ($dest, $drainer): void {
             $dest->removeListener('drain', $drainer);
         });
 
         // forward end event from source as $dest->end()
         $end = isset($options['end']) ? $options['end'] : true;
         if ($end) {
-            $source->on('end', $ender = function () use ($dest) {
+            $source->on('end', $ender = function () use ($dest): void {
                 $dest->end();
             });
-            $dest->on('close', function () use ($source, $ender) {
+            $dest->on('close', function () use ($source, $ender): void {
                 $source->removeListener('end', $ender);
             });
         }
@@ -73,7 +73,7 @@ public static function pipe(ReadableStreamInterface $source, WritableStreamInter
     public static function forwardEvents($source, $target, array $events): void
     {
         foreach ($events as $event) {
-            $source->on($event, function () use ($event, $target) {
+            $source->on($event, function () use ($event, $target): void {
                 $target->emit($event, \func_get_args());
             });
         }
diff --git a/src/WritableResourceStream.php b/src/WritableResourceStream.php
index 6140246..68b2721 100644
--- a/src/WritableResourceStream.php
+++ b/src/WritableResourceStream.php
@@ -8,24 +8,28 @@
 
 final class WritableResourceStream extends EventEmitter implements WritableStreamInterface
 {
+    /** @var resource */
     private $stream;
 
     /** @var LoopInterface */
     private $loop;
 
-    /**
-     * @var int
-     */
+    /** @var int */
     private $softLimit;
 
-    /**
-     * @var int
-     */
+    /** @var int */
     private $writeChunkSize;
 
+    /** @var bool */
     private $listening = false;
+
+    /** @var bool */
     private $writable = true;
+
+    /** @var bool */
     private $closed = false;
+
+    /** @var string */
     private $data = '';
 
     /**
@@ -119,10 +123,10 @@ public function close(): void
     }
 
     /** @internal */
-    public function handleWrite()
+    public function handleWrite(): void
     {
         $error = null;
-        \set_error_handler(function ($_, $errstr) use (&$error): bool {
+        \set_error_handler(function (int $_, string $errstr) use (&$error): bool {
             $error = $errstr;
             return true;
         });
@@ -130,6 +134,7 @@ public function handleWrite()
         if ($this->writeChunkSize === -1) {
             $sent = \fwrite($this->stream, $this->data);
         } else {
+            \assert($this->writeChunkSize >= -1);
             $sent = \fwrite($this->stream, $this->data, $this->writeChunkSize);
         }
 
@@ -151,7 +156,7 @@ public function handleWrite()
         }
 
         $exceeded = isset($this->data[$this->softLimit - 1]);
-        $this->data = (string) \substr($this->data, $sent);
+        $this->data = (string) \substr($this->data, (int) $sent);
 
         // buffer has been above limit and is now below limit
         if ($exceeded && !isset($this->data[$this->softLimit - 1])) {
diff --git a/src/WritableStreamInterface.php b/src/WritableStreamInterface.php
index 77baaad..d48536d 100644
--- a/src/WritableStreamInterface.php
+++ b/src/WritableStreamInterface.php
@@ -16,7 +16,7 @@
  *     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';
  *     });
  *     ```
@@ -37,11 +37,11 @@
  *     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();
  *         });
  *     });
@@ -64,7 +64,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;
  *     });
  *     ```
@@ -93,7 +93,7 @@
  *     The `close` event will be emitted once the stream closes (terminates).
  *
  *     ```php
- *     $stream->on('close', function () {
+ *     $stream->on('close', function (): void {
  *         echo 'CLOSED';
  *     });
  *     ```
@@ -330,7 +330,7 @@ public function end($data = null): void;
      *
      * ```php
      * $stream->end();
-     * Loop::addTimer(1.0, function () use ($stream) {
+     * Loop::addTimer(1.0, function () use ($stream): void {
      *     $stream->close();
      * });
      * ```
diff --git a/tests/CompositeStreamTest.php b/tests/CompositeStreamTest.php
index cc9ef28..6e43ad7 100644
--- a/tests/CompositeStreamTest.php
+++ b/tests/CompositeStreamTest.php
@@ -13,7 +13,7 @@
 class CompositeStreamTest extends TestCase
 {
     /** @test */
-    public function itShouldCloseReadableIfNotWritable()
+    public function itShouldCloseReadableIfNotWritable(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -39,7 +39,7 @@ public function itShouldCloseReadableIfNotWritable()
     }
 
     /** @test */
-    public function itShouldCloseWritableIfNotReadable()
+    public function itShouldCloseWritableIfNotReadable(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -61,7 +61,7 @@ public function itShouldCloseWritableIfNotReadable()
     }
 
     /** @test */
-    public function itShouldForwardWritableCallsToWritableStream()
+    public function itShouldForwardWritableCallsToWritableStream(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -87,7 +87,7 @@ public function itShouldForwardWritableCallsToWritableStream()
     }
 
     /** @test */
-    public function itShouldForwardReadableCallsToReadableStream()
+    public function itShouldForwardReadableCallsToReadableStream(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -116,7 +116,7 @@ public function itShouldForwardReadableCallsToReadableStream()
     }
 
     /** @test */
-    public function itShouldNotForwardResumeIfStreamIsNotWritable()
+    public function itShouldNotForwardResumeIfStreamIsNotWritable(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -140,7 +140,7 @@ public function itShouldNotForwardResumeIfStreamIsNotWritable()
     }
 
     /** @test */
-    public function endShouldDelegateToWritableWithData()
+    public function endShouldDelegateToWritableWithData(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -165,7 +165,7 @@ public function endShouldDelegateToWritableWithData()
     }
 
     /** @test */
-    public function closeShouldCloseBothStreams()
+    public function closeShouldCloseBothStreams(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -192,7 +192,7 @@ public function closeShouldCloseBothStreams()
     }
 
     /** @test */
-    public function itShouldForwardCloseOnlyOnce()
+    public function itShouldForwardCloseOnlyOnce(): void
     {
         $readable = new ThroughStream();
         $writable = new ThroughStream();
@@ -205,7 +205,7 @@ public function itShouldForwardCloseOnlyOnce()
     }
 
     /** @test */
-    public function itShouldForwardCloseAndRemoveAllListeners()
+    public function itShouldForwardCloseAndRemoveAllListeners(): void
     {
         $in = new ThroughStream();
 
@@ -224,7 +224,7 @@ public function itShouldForwardCloseAndRemoveAllListeners()
     }
 
     /** @test */
-    public function itShouldReceiveForwardedEvents()
+    public function itShouldReceiveForwardedEvents(): void
     {
         $readable = new ThroughStream();
         $writable = new ThroughStream();
@@ -238,7 +238,7 @@ public function itShouldReceiveForwardedEvents()
     }
 
     /** @test */
-    public function itShouldHandlePipingCorrectly()
+    public function itShouldHandlePipingCorrectly(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -263,7 +263,7 @@ public function itShouldHandlePipingCorrectly()
     }
 
     /** @test */
-    public function itShouldForwardPipeCallsToReadableStream()
+    public function itShouldForwardPipeCallsToReadableStream(): void
     {
         $readable = new ThroughStream();
 
diff --git a/tests/DuplexResourceStreamIntegrationTest.php b/tests/DuplexResourceStreamIntegrationTest.php
index a468aef..2c3d2ed 100644
--- a/tests/DuplexResourceStreamIntegrationTest.php
+++ b/tests/DuplexResourceStreamIntegrationTest.php
@@ -11,7 +11,7 @@
 
 class DuplexResourceStreamIntegrationTest extends TestCase
 {
-    public function loopProvider()
+    public function loopProvider(): \Generator
     {
         yield [
             function() {
@@ -34,7 +34,7 @@ function () {
     /**
      * @dataProvider loopProvider
      */
-    public function testBufferReadsLargeChunks($condition, $loopFactory)
+    public function testBufferReadsLargeChunks(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -42,7 +42,9 @@ public function testBufferReadsLargeChunks($condition, $loopFactory)
 
         $loop = $loopFactory();
 
-        list($sockA, $sockB) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        assert(is_array($pair));
+        [$sockA, $sockB] = $pair;
 
         $bufferSize = 4096;
         $streamA = new DuplexResourceStream($sockA, $loop, $bufferSize);
@@ -70,7 +72,7 @@ public function testBufferReadsLargeChunks($condition, $loopFactory)
     /**
      * @dataProvider loopProvider
      */
-    public function testWriteLargeChunk($condition, $loopFactory)
+    public function testWriteLargeChunk(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -78,7 +80,9 @@ public function testWriteLargeChunk($condition, $loopFactory)
 
         $loop = $loopFactory();
 
-        list($sockA, $sockB) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        assert(is_array($pair));
+        [$sockA, $sockB] = $pair;
 
         $streamA = new DuplexResourceStream($sockA, $loop);
         $streamB = new DuplexResourceStream($sockB, $loop);
@@ -110,7 +114,7 @@ public function testWriteLargeChunk($condition, $loopFactory)
     /**
      * @dataProvider loopProvider
      */
-    public function testDoesNotEmitDataIfNothingHasBeenWritten($condition, $loopFactory)
+    public function testDoesNotEmitDataIfNothingHasBeenWritten(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -118,7 +122,9 @@ public function testDoesNotEmitDataIfNothingHasBeenWritten($condition, $loopFact
 
         $loop = $loopFactory();
 
-        list($sockA, $sockB) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        assert(is_array($pair));
+        [$sockA, $sockB] = $pair;
 
         $streamA = new DuplexResourceStream($sockA, $loop);
         $streamB = new DuplexResourceStream($sockB, $loop);
@@ -138,7 +144,7 @@ public function testDoesNotEmitDataIfNothingHasBeenWritten($condition, $loopFact
     /**
      * @dataProvider loopProvider
      */
-    public function testDoesNotWriteDataIfRemoteSideFromPairHasBeenClosed($condition, $loopFactory)
+    public function testDoesNotWriteDataIfRemoteSideFromPairHasBeenClosed(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -146,7 +152,9 @@ public function testDoesNotWriteDataIfRemoteSideFromPairHasBeenClosed($condition
 
         $loop = $loopFactory();
 
-        list($sockA, $sockB) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        assert(is_array($pair));
+        [$sockA, $sockB] = $pair;
 
         $streamA = new DuplexResourceStream($sockA, $loop);
         $streamB = new DuplexResourceStream($sockB, $loop);
@@ -168,7 +176,7 @@ public function testDoesNotWriteDataIfRemoteSideFromPairHasBeenClosed($condition
     /**
      * @dataProvider loopProvider
      */
-    public function testDoesNotWriteDataIfServerSideHasBeenClosed($condition, $loopFactory)
+    public function testDoesNotWriteDataIfServerSideHasBeenClosed(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -177,9 +185,13 @@ public function testDoesNotWriteDataIfServerSideHasBeenClosed($condition, $loopF
         $loop = $loopFactory();
 
         $server = stream_socket_server('tcp://127.0.0.1:0');
+        assert(is_resource($server));
+
+        $client = stream_socket_client((string) stream_socket_get_name($server, false));
+        assert(is_resource($client));
 
-        $client = stream_socket_client(stream_socket_get_name($server, false));
         $peer = stream_socket_accept($server);
+        assert(is_resource($peer));
 
         $streamA = new DuplexResourceStream($client, $loop);
         $streamB = new DuplexResourceStream($peer, $loop);
@@ -201,7 +213,7 @@ public function testDoesNotWriteDataIfServerSideHasBeenClosed($condition, $loopF
     /**
      * @dataProvider loopProvider
      */
-    public function testDoesNotWriteDataIfClientSideHasBeenClosed($condition, $loopFactory)
+    public function testDoesNotWriteDataIfClientSideHasBeenClosed(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -210,9 +222,13 @@ public function testDoesNotWriteDataIfClientSideHasBeenClosed($condition, $loopF
         $loop = $loopFactory();
 
         $server = stream_socket_server('tcp://127.0.0.1:0');
+        assert(is_resource($server));
+
+        $client = stream_socket_client((string) stream_socket_get_name($server, false));
+        assert(is_resource($client));
 
-        $client = stream_socket_client(stream_socket_get_name($server, false));
         $peer = stream_socket_accept($server);
+        assert(is_resource($peer));
 
         $streamA = new DuplexResourceStream($peer, $loop);
         $streamB = new DuplexResourceStream($client, $loop);
@@ -234,7 +250,7 @@ public function testDoesNotWriteDataIfClientSideHasBeenClosed($condition, $loopF
     /**
      * @dataProvider loopProvider
      */
-    public function testReadsSingleChunkFromProcessPipe($condition, $loopFactory)
+    public function testReadsSingleChunkFromProcessPipe(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -242,7 +258,10 @@ public function testReadsSingleChunkFromProcessPipe($condition, $loopFactory)
 
         $loop = $loopFactory();
 
-        $stream = new ReadableResourceStream(popen('echo test', 'r'), $loop);
+        $fh = popen('echo test', 'r');
+        assert(is_resource($fh));
+
+        $stream = new ReadableResourceStream($fh, $loop);
         $stream->on('data', $this->expectCallableOnceWith("test\n"));
         $stream->on('end', $this->expectCallableOnce());
         $stream->on('error', $this->expectCallableNever());
@@ -253,7 +272,7 @@ public function testReadsSingleChunkFromProcessPipe($condition, $loopFactory)
     /**
      * @dataProvider loopProvider
      */
-    public function testReadsMultipleChunksFromProcessPipe($condition, $loopFactory)
+    public function testReadsMultipleChunksFromProcessPipe(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -261,7 +280,10 @@ public function testReadsMultipleChunksFromProcessPipe($condition, $loopFactory)
 
         $loop = $loopFactory();
 
-        $stream = new ReadableResourceStream(popen('echo a;sleep 0.1;echo b;sleep 0.1;echo c', 'r'), $loop);
+        $fh = popen('echo a;sleep 0.1;echo b;sleep 0.1;echo c', 'r');
+        assert(is_resource($fh));
+
+        $stream = new ReadableResourceStream($fh, $loop);
 
         $buffer = '';
         $stream->on('data', function ($chunk) use (&$buffer) {
@@ -279,7 +301,7 @@ public function testReadsMultipleChunksFromProcessPipe($condition, $loopFactory)
     /**
      * @dataProvider loopProvider
      */
-    public function testReadsLongChunksFromProcessPipe($condition, $loopFactory)
+    public function testReadsLongChunksFromProcessPipe(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -287,7 +309,10 @@ public function testReadsLongChunksFromProcessPipe($condition, $loopFactory)
 
         $loop = $loopFactory();
 
-        $stream = new ReadableResourceStream(popen('dd if=/dev/zero bs=12345 count=1234 2>&-', 'r'), $loop);
+        $fh = popen('dd if=/dev/zero bs=12345 count=1234 2>&-', 'r');
+        assert(is_resource($fh));
+
+        $stream = new ReadableResourceStream($fh, $loop);
 
         $bytes = 0;
         $stream->on('data', function ($chunk) use (&$bytes) {
@@ -305,7 +330,7 @@ public function testReadsLongChunksFromProcessPipe($condition, $loopFactory)
     /**
      * @dataProvider loopProvider
      */
-    public function testReadsNothingFromProcessPipeWithNoOutput($condition, $loopFactory)
+    public function testReadsNothingFromProcessPipeWithNoOutput(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
@@ -313,7 +338,10 @@ public function testReadsNothingFromProcessPipeWithNoOutput($condition, $loopFac
 
         $loop = $loopFactory();
 
-        $stream = new ReadableResourceStream(popen('true', 'r'), $loop);
+        $fh = popen('true', 'r');
+        assert(is_resource($fh));
+
+        $stream = new ReadableResourceStream($fh, $loop);
         $stream->on('data', $this->expectCallableNever());
         $stream->on('end', $this->expectCallableOnce());
         $stream->on('error', $this->expectCallableNever());
@@ -325,16 +353,20 @@ public function testReadsNothingFromProcessPipeWithNoOutput($condition, $loopFac
      * @covers React\Stream\ReadableResourceStream::handleData
      * @dataProvider loopProvider
      */
-    public function testEmptyReadShouldntFcloseStream($condition, $loopFactory)
+    public function testEmptyReadShouldntFcloseStream(callable $condition, callable $loopFactory): void
     {
         if (true !== $condition()) {
             $this->markTestSkipped('Loop implementation not available');
         }
 
         $server = stream_socket_server('tcp://127.0.0.1:0');
+        assert(is_resource($server));
+
+        $client = stream_socket_client((string) stream_socket_get_name($server, false));
+        assert(is_resource($client));
 
-        $client = stream_socket_client(stream_socket_get_name($server, false));
         $stream = stream_socket_accept($server);
+        assert(is_resource($stream));
 
 
         // add a filter which returns an error when encountering an 'a' when reading
@@ -358,7 +390,7 @@ public function testEmptyReadShouldntFcloseStream($condition, $loopFactory)
         fclose($server);
     }
 
-    private function loopTick(LoopInterface $loop)
+    private function loopTick(LoopInterface $loop): void
     {
         $loop->addTimer(0, function () use ($loop) {
             $loop->stop();
diff --git a/tests/DuplexResourceStreamTest.php b/tests/DuplexResourceStreamTest.php
index 5b18553..c1076a6 100644
--- a/tests/DuplexResourceStreamTest.php
+++ b/tests/DuplexResourceStreamTest.php
@@ -2,6 +2,7 @@
 
 namespace React\Tests\Stream;
 
+use PHPUnit\Framework\MockObject\MockObject;
 use React\EventLoop\LoopInterface;
 use React\Stream\DuplexResourceStream;
 use React\Stream\WritableResourceStream;
@@ -14,17 +15,20 @@ class DuplexResourceStreamTest extends TestCase
      * @covers React\Stream\DuplexResourceStream::__construct
      * @doesNotPerformAssertions
      */
-    public function testConstructor()
+    public function testConstructor(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         new DuplexResourceStream($stream, $loop);
     }
 
-    public function testConstructWithoutLoopAssignsLoopAutomatically()
+    public function testConstructWithoutLoopAssignsLoopAutomatically(): void
     {
         $resource = fopen('php://temp', 'r+');
+        assert(is_resource($resource));
 
         $stream = new DuplexResourceStream($resource);
 
@@ -39,11 +43,12 @@ public function testConstructWithoutLoopAssignsLoopAutomatically()
      * @covers React\Stream\DuplexResourceStream::__construct
      * @doesNotPerformAssertions
      */
-    public function testConstructorWithExcessiveMode()
+    public function testConstructorWithExcessiveMode(): void
     {
         // excessive flags are ignored for temp streams, so we have to use a file stream
-        $name = tempnam(sys_get_temp_dir(), 'test');
-        $stream = @fopen($name, 'r+eANYTHING');
+        $name = (string) tempnam(sys_get_temp_dir(), 'test');
+        $stream = fopen($name, 'r+eANYTHING');
+        assert(is_resource($stream));
         unlink($name);
 
         $loop = $this->createLoopMock();
@@ -54,7 +59,7 @@ public function testConstructorWithExcessiveMode()
     /**
      * @covers React\Stream\DuplexResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionOnInvalidStream()
+    public function testConstructorThrowsExceptionOnInvalidStream(): void
     {
         $loop = $this->createLoopMock();
 
@@ -65,7 +70,7 @@ public function testConstructorThrowsExceptionOnInvalidStream()
     /**
      * @covers React\Stream\DuplexResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionOnWriteOnlyStream()
+    public function testConstructorThrowsExceptionOnWriteOnlyStream(): void
     {
         $loop = $this->createLoopMock();
 
@@ -76,11 +81,12 @@ public function testConstructorThrowsExceptionOnWriteOnlyStream()
     /**
      * @covers React\Stream\DuplexResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionOnWriteOnlyStreamWithExcessiveMode()
+    public function testConstructorThrowsExceptionOnWriteOnlyStreamWithExcessiveMode(): void
     {
         // excessive flags are ignored for temp streams, so we have to use a file stream
-        $name = tempnam(sys_get_temp_dir(), 'test');
+        $name = (string) tempnam(sys_get_temp_dir(), 'test');
         $stream = fopen($name, 'weANYTHING');
+        assert(is_resource($stream));
         unlink($name);
 
         $loop = $this->createLoopMock();
@@ -91,13 +97,15 @@ public function testConstructorThrowsExceptionOnWriteOnlyStreamWithExcessiveMode
     /**
      * @covers React\Stream\DuplexResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlocking()
+    public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlocking(): void
     {
         if (!in_array('blocking', stream_get_wrappers())) {
             stream_wrapper_register('blocking', 'React\Tests\Stream\EnforceBlockingWrapper');
         }
 
         $stream = fopen('blocking://test', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $this->expectException(\RuntimeException::class);
@@ -108,9 +116,11 @@ public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlocking(
      * @covers React\Stream\DuplexResourceStream::__construct
      * @doesNotPerformAssertions
      */
-    public function testConstructorAcceptsBuffer()
+    public function testConstructorAcceptsBuffer(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = $this->createMock(WritableStreamInterface::class);
@@ -122,13 +132,15 @@ public function testConstructorAcceptsBuffer()
     /**
      * @covers React\Stream\DuplexResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlockingWithBufferGiven()
+    public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlockingWithBufferGiven(): void
     {
         if (!in_array('blocking', stream_get_wrappers())) {
             stream_wrapper_register('blocking', 'React\Tests\Stream\EnforceBlockingWrapper');
         }
 
         $stream = fopen('blocking://test', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
@@ -138,9 +150,11 @@ public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlockingW
         new DuplexResourceStream($stream, $loop, null, $buffer);
     }
 
-    public function testCloseShouldEmitCloseEvent()
+    public function testCloseShouldEmitCloseEvent(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new DuplexResourceStream($stream, $loop);
@@ -152,9 +166,11 @@ public function testCloseShouldEmitCloseEvent()
         $this->assertFalse($conn->isReadable());
     }
 
-    public function testEndShouldEndBuffer()
+    public function testEndShouldEndBuffer(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = $this->createMock(WritableStreamInterface::class);
@@ -166,9 +182,11 @@ public function testEndShouldEndBuffer()
     }
 
 
-    public function testEndAfterCloseIsNoOp()
+    public function testEndAfterCloseIsNoOp(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = $this->createMock(WritableStreamInterface::class);
@@ -184,9 +202,11 @@ public function testEndAfterCloseIsNoOp()
      * @covers React\Stream\DuplexResourceStream::__construct
      * @covers React\Stream\DuplexResourceStream::handleData
      */
-    public function testDataEvent()
+    public function testDataEvent(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $capturedData = null;
@@ -207,9 +227,11 @@ public function testDataEvent()
      * @covers React\Stream\DuplexResourceStream::__construct
      * @covers React\Stream\DuplexResourceStream::handleData
      */
-    public function testDataEventDoesEmitOneChunkMatchingBufferSize()
+    public function testDataEventDoesEmitOneChunkMatchingBufferSize(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $capturedData = null;
@@ -232,9 +254,11 @@ public function testDataEventDoesEmitOneChunkMatchingBufferSize()
      * @covers React\Stream\DuplexResourceStream::__construct
      * @covers React\Stream\DuplexResourceStream::handleData
      */
-    public function testDataEventDoesEmitOneChunkUntilStreamEndsWhenBufferSizeIsInfinite()
+    public function testDataEventDoesEmitOneChunkUntilStreamEndsWhenBufferSizeIsInfinite(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $capturedData = null;
@@ -257,9 +281,11 @@ public function testDataEventDoesEmitOneChunkUntilStreamEndsWhenBufferSizeIsInfi
     /**
      * @covers React\Stream\DuplexResourceStream::handleData
      */
-    public function testEmptyStreamShouldNotEmitData()
+    public function testEmptyStreamShouldNotEmitData(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new DuplexResourceStream($stream, $loop);
@@ -271,9 +297,11 @@ public function testEmptyStreamShouldNotEmitData()
     /**
      * @covers React\Stream\DuplexResourceStream::write
      */
-    public function testWrite()
+    public function testWrite(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createWriteableLoopMock();
 
         $conn = new DuplexResourceStream($stream, $loop);
@@ -288,9 +316,11 @@ public function testWrite()
      * @covers React\Stream\DuplexResourceStream::isReadable
      * @covers React\Stream\DuplexResourceStream::isWritable
      */
-    public function testEnd()
+    public function testEnd(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new DuplexResourceStream($stream, $loop);
@@ -304,9 +334,11 @@ public function testEnd()
     /**
      * @covers React\Stream\DuplexResourceStream::end
      */
-    public function testEndRemovesReadStreamFromLoop()
+    public function testEndRemovesReadStreamFromLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
         $loop->expects($this->once())->method('removeReadStream')->with($stream);
@@ -318,9 +350,11 @@ public function testEndRemovesReadStreamFromLoop()
     /**
      * @covers React\Stream\DuplexResourceStream::pause
      */
-    public function testPauseRemovesReadStreamFromLoop()
+    public function testPauseRemovesReadStreamFromLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
         $loop->expects($this->once())->method('removeReadStream')->with($stream);
@@ -333,9 +367,11 @@ public function testPauseRemovesReadStreamFromLoop()
     /**
      * @covers React\Stream\DuplexResourceStream::pause
      */
-    public function testResumeDoesAddStreamToLoopOnlyOnce()
+    public function testResumeDoesAddStreamToLoopOnlyOnce(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
 
@@ -347,9 +383,11 @@ public function testResumeDoesAddStreamToLoopOnlyOnce()
     /**
      * @covers React\Stream\DuplexResourceStream::close
      */
-    public function testCloseRemovesReadStreamFromLoop()
+    public function testCloseRemovesReadStreamFromLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
         $loop->expects($this->once())->method('removeReadStream')->with($stream);
@@ -361,9 +399,11 @@ public function testCloseRemovesReadStreamFromLoop()
     /**
      * @covers React\Stream\DuplexResourceStream::close
      */
-    public function testCloseAfterPauseRemovesReadStreamFromLoopOnlyOnce()
+    public function testCloseAfterPauseRemovesReadStreamFromLoopOnlyOnce(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
         $loop->expects($this->once())->method('removeReadStream')->with($stream);
@@ -376,9 +416,11 @@ public function testCloseAfterPauseRemovesReadStreamFromLoopOnlyOnce()
     /**
      * @covers React\Stream\DuplexResourceStream::close
      */
-    public function testResumeAfterCloseDoesAddReadStreamToLoopOnlyOnce()
+    public function testResumeAfterCloseDoesAddReadStreamToLoopOnlyOnce(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
 
@@ -387,10 +429,12 @@ public function testResumeAfterCloseDoesAddReadStreamToLoopOnlyOnce()
         $conn->resume();
     }
 
-    public function testEndedStreamsShouldNotWrite()
+    public function testEndedStreamsShouldNotWrite(): void
     {
-        $file = tempnam(sys_get_temp_dir(), 'reactphptest_');
+        $file = (string) tempnam(sys_get_temp_dir(), 'reactphptest_');
         $stream = fopen($file, 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createWriteableLoopMock();
 
         $conn = new DuplexResourceStream($stream, $loop);
@@ -398,7 +442,9 @@ public function testEndedStreamsShouldNotWrite()
         $conn->end();
 
         $res = $conn->write("bar\n");
+
         $stream = fopen($file, 'r');
+        assert(is_resource($stream));
 
         $this->assertSame("foo\n", fgets($stream));
         $this->assertFalse($res);
@@ -406,9 +452,11 @@ public function testEndedStreamsShouldNotWrite()
         unlink($file);
     }
 
-    public function testPipeShouldReturnDestination()
+    public function testPipeShouldReturnDestination(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new DuplexResourceStream($stream, $loop);
@@ -418,9 +466,11 @@ public function testPipeShouldReturnDestination()
         $this->assertSame($dest, $conn->pipe($dest));
     }
 
-    public function testBufferEventsShouldBubbleUp()
+    public function testBufferEventsShouldBubbleUp(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop);
@@ -436,9 +486,11 @@ public function testBufferEventsShouldBubbleUp()
     /**
      * @covers React\Stream\DuplexResourceStream::handleData
      */
-    public function testClosingStreamInDataEventShouldNotTriggerError()
+    public function testClosingStreamInDataEventShouldNotTriggerError(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new DuplexResourceStream($stream, $loop);
@@ -456,9 +508,10 @@ public function testClosingStreamInDataEventShouldNotTriggerError()
     /**
      * @covers React\Stream\DuplexResourceStream::handleData
      */
-    public function testDataFiltered()
+    public function testDataFiltered(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
 
         // add a filter which removes every 'a' when reading
         filter_append($stream, function ($chunk) {
@@ -484,9 +537,10 @@ public function testDataFiltered()
     /**
      * @covers React\Stream\DuplexResourceStream::handleData
      */
-    public function testDataErrorShouldEmitErrorAndClose()
+    public function testDataErrorShouldEmitErrorAndClose(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
 
         // add a filter which returns an error when encountering an 'a' when reading
         filter_append($stream, function ($chunk) {
@@ -509,7 +563,7 @@ public function testDataErrorShouldEmitErrorAndClose()
         $conn->handleData($stream);
     }
 
-    private function createWriteableLoopMock()
+    private function createWriteableLoopMock(): LoopInterface
     {
         $loop = $this->createLoopMock();
         $loop
@@ -522,8 +576,10 @@ private function createWriteableLoopMock()
         return $loop;
     }
 
-    private function createLoopMock()
+    /** @return LoopInterface&MockObject */
+    private function createLoopMock(): MockObject
     {
+        /** @var MockObject&LoopInterface */
         return $this->createMock(LoopInterface::class);
     }
 }
diff --git a/tests/EnforceBlockingWrapper.php b/tests/EnforceBlockingWrapper.php
index a171b41..cab433b 100644
--- a/tests/EnforceBlockingWrapper.php
+++ b/tests/EnforceBlockingWrapper.php
@@ -9,24 +9,25 @@
  */
 class EnforceBlockingWrapper
 {
+    /** @var resource */
     public $context;
 
-    public function stream_open($path, $mode, $options, &$opened_path)
+    public function stream_open(string $path, string $mode, int $options, ?string &$opened_path): bool
     {
         return true;
     }
 
-    public function stream_cast($cast_as)
+    public function stream_cast(int $cast_as): bool
     {
         return false;
     }
 
-    public function stream_eof()
+    public function stream_eof(): bool
     {
         return false;
     }
 
-    public function stream_set_option($option, $arg1, $arg2)
+    public function stream_set_option(int $option, int $arg1, ?int $arg2): bool
     {
         if ($option === STREAM_OPTION_BLOCKING) {
             return false;
diff --git a/tests/FunctionalInternetTest.php b/tests/FunctionalInternetTest.php
index 5113e7b..f56eefc 100644
--- a/tests/FunctionalInternetTest.php
+++ b/tests/FunctionalInternetTest.php
@@ -12,10 +12,11 @@
  */
 class FunctionalInternetTest extends TestCase
 {
-    public function testUploadKilobytePlain()
+    public function testUploadKilobytePlain(): void
     {
         $size = 1000;
         $stream = stream_socket_client('tcp://httpbin.org:80');
+        assert(is_resource($stream));
 
         $loop = Factory::create();
         $stream = new DuplexResourceStream($stream, $loop);
@@ -34,10 +35,11 @@ public function testUploadKilobytePlain()
         $this->assertNotEquals('', $buffer);
     }
 
-    public function testUploadBiggerBlockPlain()
+    public function testUploadBiggerBlockPlain(): void
     {
         $size = 50 * 1000;
         $stream = stream_socket_client('tcp://httpbin.org:80');
+        assert(is_resource($stream));
 
         $loop = Factory::create();
         $stream = new DuplexResourceStream($stream, $loop);
@@ -56,10 +58,11 @@ public function testUploadBiggerBlockPlain()
         $this->assertNotEquals('', $buffer);
     }
 
-    public function testUploadKilobyteSecure()
+    public function testUploadKilobyteSecure(): void
     {
         $size = 1000;
         $stream = stream_socket_client('ssl://httpbin.org:443');
+        assert(is_resource($stream));
 
         $loop = Factory::create();
         $stream = new DuplexResourceStream($stream, $loop);
@@ -78,7 +81,7 @@ public function testUploadKilobyteSecure()
         $this->assertNotEquals('', $buffer);
     }
 
-    public function testUploadBiggerBlockSecure()
+    public function testUploadBiggerBlockSecure(): void
     {
         // A few dozen kilobytes should be enough to verify this works.
         // Underlying buffer sizes are platform-specific, so let's increase this
@@ -86,6 +89,7 @@ public function testUploadBiggerBlockSecure()
         $size = 136 * 1000;
 
         $stream = stream_socket_client('ssl://httpbin.org:443');
+        assert(is_resource($stream));
 
         // PHP < 7.1.4 suffers from a bug when writing big chunks of data over
         // TLS streams at once.
@@ -114,7 +118,7 @@ public function testUploadBiggerBlockSecure()
         $this->assertNotEquals('', $buffer);
     }
 
-    private function awaitStreamClose(DuplexResourceStream $stream, LoopInterface $loop, $timeout = 10.0)
+    private function awaitStreamClose(DuplexResourceStream $stream, LoopInterface $loop, float $timeout = 10.0): void
     {
         $stream->on('close', function () use ($loop) {
             $loop->stop();
diff --git a/tests/ReadableResourceStreamTest.php b/tests/ReadableResourceStreamTest.php
index 365ac0d..3f7c144 100644
--- a/tests/ReadableResourceStreamTest.php
+++ b/tests/ReadableResourceStreamTest.php
@@ -2,6 +2,7 @@
 
 namespace React\Tests\Stream;
 
+use PHPUnit\Framework\MockObject\MockObject;
 use React\EventLoop\LoopInterface;
 use React\Stream\ReadableResourceStream;
 use React\Stream\WritableStreamInterface;
@@ -13,17 +14,20 @@ class ReadableResourceStreamTest extends TestCase
      * @covers React\Stream\ReadableResourceStream::__construct
      * @doesNotPerformAssertions
      */
-    public function testConstructor()
+    public function testConstructor(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         new ReadableResourceStream($stream, $loop);
     }
 
-    public function testConstructWithoutLoopAssignsLoopAutomatically()
+    public function testConstructWithoutLoopAssignsLoopAutomatically(): void
     {
         $resource = fopen('php://temp', 'r+');
+        assert(is_resource($resource));
 
         $stream = new ReadableResourceStream($resource);
 
@@ -38,11 +42,12 @@ public function testConstructWithoutLoopAssignsLoopAutomatically()
      * @covers React\Stream\ReadableResourceStream::__construct
      * @doesNotPerformAssertions
      */
-    public function testConstructorWithExcessiveMode()
+    public function testConstructorWithExcessiveMode(): void
     {
         // excessive flags are ignored for temp streams, so we have to use a file stream
-        $name = tempnam(sys_get_temp_dir(), 'test');
-        $stream = @fopen($name, 'r+eANYTHING');
+        $name = (string) tempnam(sys_get_temp_dir(), 'test');
+        $stream = fopen($name, 'r+eANYTHING');
+        assert(is_resource($stream));
         unlink($name);
 
         $loop = $this->createLoopMock();
@@ -53,7 +58,7 @@ public function testConstructorWithExcessiveMode()
     /**
      * @covers React\Stream\ReadableResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionOnInvalidStream()
+    public function testConstructorThrowsExceptionOnInvalidStream(): void
     {
         $loop = $this->createLoopMock();
 
@@ -64,7 +69,7 @@ public function testConstructorThrowsExceptionOnInvalidStream()
     /**
      * @covers React\Stream\ReadableResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionOnWriteOnlyStream()
+    public function testConstructorThrowsExceptionOnWriteOnlyStream(): void
     {
         $loop = $this->createLoopMock();
 
@@ -75,11 +80,12 @@ public function testConstructorThrowsExceptionOnWriteOnlyStream()
     /**
      * @covers React\Stream\ReadableResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionOnWriteOnlyStreamWithExcessiveMode()
+    public function testConstructorThrowsExceptionOnWriteOnlyStreamWithExcessiveMode(): void
     {
         // excessive flags are ignored for temp streams, so we have to use a file stream
-        $name = tempnam(sys_get_temp_dir(), 'test');
+        $name = (string) tempnam(sys_get_temp_dir(), 'test');
         $stream = fopen($name, 'weANYTHING');
+        assert(is_resource($stream));
         unlink($name);
 
         $loop = $this->createLoopMock();
@@ -90,23 +96,26 @@ public function testConstructorThrowsExceptionOnWriteOnlyStreamWithExcessiveMode
     /**
      * @covers React\Stream\ReadableResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlocking()
+    public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlocking(): void
     {
         if (!in_array('blocking', stream_get_wrappers())) {
             stream_wrapper_register('blocking', 'React\Tests\Stream\EnforceBlockingWrapper');
         }
 
         $stream = fopen('blocking://test', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $this->expectException(\RuntimeException::class);
         new ReadableResourceStream($stream, $loop);
     }
 
-
-    public function testCloseShouldEmitCloseEvent()
+    public function testCloseShouldEmitCloseEvent(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new ReadableResourceStream($stream, $loop);
@@ -117,9 +126,11 @@ public function testCloseShouldEmitCloseEvent()
         $this->assertFalse($conn->isReadable());
     }
 
-    public function testCloseTwiceShouldEmitCloseEventOnce()
+    public function testCloseTwiceShouldEmitCloseEventOnce(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new ReadableResourceStream($stream, $loop);
@@ -133,9 +144,11 @@ public function testCloseTwiceShouldEmitCloseEventOnce()
      * @covers React\Stream\ReadableResourceStream::__construct
      * @covers React\Stream\ReadableResourceStream::handleData
      */
-    public function testDataEvent()
+    public function testDataEvent(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $capturedData = null;
@@ -156,9 +169,11 @@ public function testDataEvent()
      * @covers React\Stream\ReadableResourceStream::__construct
      * @covers React\Stream\ReadableResourceStream::handleData
      */
-    public function testDataEventDoesEmitOneChunkMatchingBufferSize()
+    public function testDataEventDoesEmitOneChunkMatchingBufferSize(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $capturedData = null;
@@ -181,9 +196,11 @@ public function testDataEventDoesEmitOneChunkMatchingBufferSize()
      * @covers React\Stream\ReadableResourceStream::__construct
      * @covers React\Stream\ReadableResourceStream::handleData
      */
-    public function testDataEventDoesEmitOneChunkUntilStreamEndsWhenBufferSizeIsInfinite()
+    public function testDataEventDoesEmitOneChunkUntilStreamEndsWhenBufferSizeIsInfinite(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $capturedData = null;
@@ -206,9 +223,11 @@ public function testDataEventDoesEmitOneChunkUntilStreamEndsWhenBufferSizeIsInfi
     /**
      * @covers React\Stream\ReadableResourceStream::handleData
      */
-    public function testEmptyStreamShouldNotEmitData()
+    public function testEmptyStreamShouldNotEmitData(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new ReadableResourceStream($stream, $loop);
@@ -217,9 +236,11 @@ public function testEmptyStreamShouldNotEmitData()
         $conn->handleData();
     }
 
-    public function testPipeShouldReturnDestination()
+    public function testPipeShouldReturnDestination(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new ReadableResourceStream($stream, $loop);
@@ -232,9 +253,11 @@ public function testPipeShouldReturnDestination()
     /**
      * @covers React\Stream\ReadableResourceStream::handleData
      */
-    public function testClosingStreamInDataEventShouldNotTriggerError()
+    public function testClosingStreamInDataEventShouldNotTriggerError(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $conn = new ReadableResourceStream($stream, $loop);
@@ -252,9 +275,11 @@ public function testClosingStreamInDataEventShouldNotTriggerError()
     /**
      * @covers React\Stream\ReadableResourceStream::pause
      */
-    public function testPauseRemovesReadStreamFromLoop()
+    public function testPauseRemovesReadStreamFromLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
         $loop->expects($this->once())->method('removeReadStream')->with($stream);
@@ -267,9 +292,11 @@ public function testPauseRemovesReadStreamFromLoop()
     /**
      * @covers React\Stream\ReadableResourceStream::pause
      */
-    public function testResumeDoesAddStreamToLoopOnlyOnce()
+    public function testResumeDoesAddStreamToLoopOnlyOnce(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
 
@@ -281,9 +308,11 @@ public function testResumeDoesAddStreamToLoopOnlyOnce()
     /**
      * @covers React\Stream\ReadableResourceStream::close
      */
-    public function testCloseRemovesReadStreamFromLoop()
+    public function testCloseRemovesReadStreamFromLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
         $loop->expects($this->once())->method('removeReadStream')->with($stream);
@@ -295,9 +324,11 @@ public function testCloseRemovesReadStreamFromLoop()
     /**
      * @covers React\Stream\ReadableResourceStream::close
      */
-    public function testCloseAfterPauseRemovesReadStreamFromLoopOnce()
+    public function testCloseAfterPauseRemovesReadStreamFromLoopOnce(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
         $loop->expects($this->once())->method('removeReadStream')->with($stream);
@@ -310,9 +341,11 @@ public function testCloseAfterPauseRemovesReadStreamFromLoopOnce()
     /**
      * @covers React\Stream\ReadableResourceStream::close
      */
-    public function testResumeAfterCloseDoesAddReadStreamToLoopOnlyOnce()
+    public function testResumeAfterCloseDoesAddReadStreamToLoopOnlyOnce(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addReadStream')->with($stream);
 
@@ -324,9 +357,10 @@ public function testResumeAfterCloseDoesAddReadStreamToLoopOnlyOnce()
     /**
      * @covers React\Stream\ReadableResourceStream::handleData
      */
-    public function testDataFiltered()
+    public function testDataFiltered(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
 
         // add a filter which removes every 'a' when reading
         filter_append($stream, function ($chunk) {
@@ -352,9 +386,10 @@ public function testDataFiltered()
     /**
      * @covers React\Stream\ReadableResourceStream::handleData
      */
-    public function testDataErrorShouldEmitErrorAndClose()
+    public function testDataErrorShouldEmitErrorAndClose(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
 
         // add a filter which returns an error when encountering an 'a' when reading
         filter_append($stream, function ($chunk) {
@@ -380,9 +415,12 @@ public function testDataErrorShouldEmitErrorAndClose()
     /**
      * @covers React\Stream\ReadableResourceStream::handleData
      */
-    public function testEmptyReadShouldntFcloseStream()
+    public function testEmptyReadShouldntFcloseStream(): void
     {
-        list($stream, $_) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        assert(is_array($pair));
+        [$stream, $_] = $pair;
+
         $loop = $this->createLoopMock();
 
         $conn = new ReadableResourceStream($stream, $loop);
@@ -396,8 +434,10 @@ public function testEmptyReadShouldntFcloseStream()
         fclose($_);
     }
 
-    private function createLoopMock()
+    /** @return MockObject&LoopInterface */
+    private function createLoopMock(): MockObject
     {
+        /** @var MockObject&LoopInterface */
         return $this->createMock(LoopInterface::class);
     }
 }
diff --git a/tests/Stub/ReadableStreamStub.php b/tests/Stub/ReadableStreamStub.php
index 669059c..98d0240 100644
--- a/tests/Stub/ReadableStreamStub.php
+++ b/tests/Stub/ReadableStreamStub.php
@@ -9,7 +9,10 @@
 
 class ReadableStreamStub extends EventEmitter implements ReadableStreamInterface
 {
+    /** @var bool */
     public $readable = true;
+
+    /** @var bool */
     public $paused = false;
 
     public function isReadable(): bool
@@ -17,20 +20,25 @@ public function isReadable(): bool
         return true;
     }
 
-    // trigger data event
-    public function write($data)
+    /**
+     * trigger data event
+     *
+     * @param mixed $data
+     * @return void
+     */
+    public function write($data): void
     {
         $this->emit('data', [$data]);
     }
 
     // trigger error event
-    public function error($error)
+    public function error(\Exception $error): void
     {
         $this->emit('error', [$error]);
     }
 
     // trigger end event
-    public function end()
+    public function end(): void
     {
         $this->emit('end', []);
     }
diff --git a/tests/TestCase.php b/tests/TestCase.php
index b161c46..95bc3f1 100644
--- a/tests/TestCase.php
+++ b/tests/TestCase.php
@@ -2,11 +2,12 @@
 
 namespace React\Tests\Stream;
 
+use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase as BaseTestCase;
 
 class TestCase extends BaseTestCase
 {
-    protected function expectCallableOnce()
+    protected function expectCallableOnce(): callable
     {
         $mock = $this->createCallableMock();
         $mock
@@ -16,7 +17,8 @@ protected function expectCallableOnce()
         return $mock;
     }
 
-    protected function expectCallableOnceWith($value)
+    /** @param mixed $value */
+    protected function expectCallableOnceWith($value): callable
     {
         $callback = $this->createCallableMock();
         $callback
@@ -27,7 +29,7 @@ protected function expectCallableOnceWith($value)
         return $callback;
     }
 
-    protected function expectCallableNever()
+    protected function expectCallableNever(): callable
     {
         $mock = $this->createCallableMock();
         $mock
@@ -37,15 +39,19 @@ protected function expectCallableNever()
         return $mock;
     }
 
-    protected function createCallableMock()
+    /** @return MockObject&callable */
+    protected function createCallableMock(): MockObject
     {
         $builder = $this->getMockBuilder(\stdClass::class);
         if (method_exists($builder, 'addMethods')) {
             // PHPUnit 9+
-            return $builder->addMethods(['__invoke'])->getMock();
+            $mock = $builder->addMethods(['__invoke'])->getMock();
         } else {
             // legacy PHPUnit
-            return $builder->setMethods(['__invoke'])->getMock();
+            $mock = $builder->setMethods(['__invoke'])->getMock();
         }
+        assert($mock instanceof MockObject && is_callable($mock));
+
+        return $mock;
     }
 }
diff --git a/tests/ThroughStreamTest.php b/tests/ThroughStreamTest.php
index ddd570f..a4b49b0 100644
--- a/tests/ThroughStreamTest.php
+++ b/tests/ThroughStreamTest.php
@@ -11,7 +11,7 @@
 class ThroughStreamTest extends TestCase
 {
     /** @test */
-    public function itShouldReturnTrueForAnyDataWrittenToIt()
+    public function itShouldReturnTrueForAnyDataWrittenToIt(): void
     {
         $through = new ThroughStream();
         $ret = $through->write('foo');
@@ -20,7 +20,7 @@ public function itShouldReturnTrueForAnyDataWrittenToIt()
     }
 
     /** @test */
-    public function itShouldEmitAnyDataWrittenToIt()
+    public function itShouldEmitAnyDataWrittenToIt(): void
     {
         $through = new ThroughStream();
         $through->on('data', $this->expectCallableOnceWith('foo'));
@@ -28,7 +28,7 @@ public function itShouldEmitAnyDataWrittenToIt()
     }
 
     /** @test */
-    public function itShouldEmitAnyDataWrittenToItPassedThruFunction()
+    public function itShouldEmitAnyDataWrittenToItPassedThruFunction(): void
     {
         $through = new ThroughStream('strtoupper');
         $through->on('data', $this->expectCallableOnceWith('FOO'));
@@ -36,7 +36,7 @@ public function itShouldEmitAnyDataWrittenToItPassedThruFunction()
     }
 
     /** @test */
-    public function itShouldEmitAnyDataWrittenToItPassedThruCallback()
+    public function itShouldEmitAnyDataWrittenToItPassedThruCallback(): void
     {
         $through = new ThroughStream('strtoupper');
         $through->on('data', $this->expectCallableOnceWith('FOO'));
@@ -44,7 +44,7 @@ public function itShouldEmitAnyDataWrittenToItPassedThruCallback()
     }
 
     /** @test */
-    public function itShouldEmitErrorAndCloseIfCallbackThrowsException()
+    public function itShouldEmitErrorAndCloseIfCallbackThrowsException(): void
     {
         $through = new ThroughStream(function () {
             throw new \RuntimeException();
@@ -61,7 +61,7 @@ public function itShouldEmitErrorAndCloseIfCallbackThrowsException()
     }
 
     /** @test */
-    public function itShouldEmitErrorAndCloseIfCallbackThrowsExceptionOnEnd()
+    public function itShouldEmitErrorAndCloseIfCallbackThrowsExceptionOnEnd(): void
     {
         $through = new ThroughStream(function () {
             throw new \RuntimeException();
@@ -78,7 +78,7 @@ public function itShouldEmitErrorAndCloseIfCallbackThrowsExceptionOnEnd()
     }
 
     /** @test */
-    public function itShouldReturnFalseForAnyDataWrittenToItWhenPaused()
+    public function itShouldReturnFalseForAnyDataWrittenToItWhenPaused(): void
     {
         $through = new ThroughStream();
         $through->pause();
@@ -88,7 +88,7 @@ public function itShouldReturnFalseForAnyDataWrittenToItWhenPaused()
     }
 
     /** @test */
-    public function itShouldReturnFalseForAnyDataWrittenToItWhenDataEventEndsStream()
+    public function itShouldReturnFalseForAnyDataWrittenToItWhenDataEventEndsStream(): void
     {
         $through = new ThroughStream();
         $through->on('data', function () use ($through) {
@@ -100,7 +100,7 @@ public function itShouldReturnFalseForAnyDataWrittenToItWhenDataEventEndsStream(
     }
 
     /** @test */
-    public function itShouldReturnFalseForAnyDataWrittenToItWhenDataEventClosesStream()
+    public function itShouldReturnFalseForAnyDataWrittenToItWhenDataEventClosesStream(): void
     {
         $through = new ThroughStream();
         $through->on('data', function () use ($through) {
@@ -112,7 +112,7 @@ public function itShouldReturnFalseForAnyDataWrittenToItWhenDataEventClosesStrea
     }
 
     /** @test */
-    public function itShouldEmitDrainOnResumeAfterReturnFalseForAnyDataWrittenToItWhenPaused()
+    public function itShouldEmitDrainOnResumeAfterReturnFalseForAnyDataWrittenToItWhenPaused(): void
     {
         $through = new ThroughStream();
         $through->pause();
@@ -123,7 +123,7 @@ public function itShouldEmitDrainOnResumeAfterReturnFalseForAnyDataWrittenToItWh
     }
 
     /** @test */
-    public function itShouldNotEmitDrainOnResumeAfterClose()
+    public function itShouldNotEmitDrainOnResumeAfterClose(): void
     {
         $through = new ThroughStream();
         $through->close();
@@ -133,7 +133,7 @@ public function itShouldNotEmitDrainOnResumeAfterClose()
     }
 
     /** @test */
-    public function itShouldNotEmitDrainOnResumeAfterReturnFalseForAnyDataWrittenThatCausesStreamToClose()
+    public function itShouldNotEmitDrainOnResumeAfterReturnFalseForAnyDataWrittenThatCausesStreamToClose(): void
     {
         $through = new ThroughStream();
         $through->on('data', function () use ($through) { $through->close(); });
@@ -144,7 +144,7 @@ public function itShouldNotEmitDrainOnResumeAfterReturnFalseForAnyDataWrittenTha
     }
 
     /** @test */
-    public function itShouldReturnFalseForAnyDataWrittenToItAfterPausingFromDrainEvent()
+    public function itShouldReturnFalseForAnyDataWrittenToItAfterPausingFromDrainEvent(): void
     {
         $through = new ThroughStream();
         $through->pause();
@@ -157,7 +157,7 @@ public function itShouldReturnFalseForAnyDataWrittenToItAfterPausingFromDrainEve
     }
 
     /** @test */
-    public function itShouldReturnTrueForAnyDataWrittenToItWhenResumedAfterPause()
+    public function itShouldReturnTrueForAnyDataWrittenToItWhenResumedAfterPause(): void
     {
         $through = new ThroughStream();
         $through->on('drain', $this->expectCallableNever());
@@ -169,7 +169,7 @@ public function itShouldReturnTrueForAnyDataWrittenToItWhenResumedAfterPause()
     }
 
     /** @test */
-    public function pipingStuffIntoItShouldWork()
+    public function pipingStuffIntoItShouldWork(): void
     {
         $readable = new ThroughStream();
 
@@ -181,7 +181,7 @@ public function pipingStuffIntoItShouldWork()
     }
 
     /** @test */
-    public function endShouldEmitEndAndClose()
+    public function endShouldEmitEndAndClose(): void
     {
         $through = new ThroughStream();
         $through->on('data', $this->expectCallableNever());
@@ -191,7 +191,7 @@ public function endShouldEmitEndAndClose()
     }
 
     /** @test */
-    public function endShouldCloseTheStream()
+    public function endShouldCloseTheStream(): void
     {
         $through = new ThroughStream();
         $through->on('data', $this->expectCallableNever());
@@ -202,7 +202,7 @@ public function endShouldCloseTheStream()
     }
 
     /** @test */
-    public function endShouldWriteDataBeforeClosing()
+    public function endShouldWriteDataBeforeClosing(): void
     {
         $through = new ThroughStream();
         $through->on('data', $this->expectCallableOnceWith('foo'));
@@ -213,7 +213,7 @@ public function endShouldWriteDataBeforeClosing()
     }
 
     /** @test */
-    public function endTwiceShouldOnlyEmitOnce()
+    public function endTwiceShouldOnlyEmitOnce(): void
     {
         $through = new ThroughStream();
         $through->on('data', $this->expectCallableOnceWith('first'));
@@ -222,7 +222,7 @@ public function endTwiceShouldOnlyEmitOnce()
     }
 
     /** @test */
-    public function writeAfterEndShouldReturnFalse()
+    public function writeAfterEndShouldReturnFalse(): void
     {
         $through = new ThroughStream();
         $through->on('data', $this->expectCallableNever());
@@ -232,7 +232,7 @@ public function writeAfterEndShouldReturnFalse()
     }
 
     /** @test */
-    public function writeDataWillCloseStreamShouldReturnFalse()
+    public function writeDataWillCloseStreamShouldReturnFalse(): void
     {
         $through = new ThroughStream();
         $through->on('data', [$through, 'close']);
@@ -241,7 +241,7 @@ public function writeDataWillCloseStreamShouldReturnFalse()
     }
 
     /** @test */
-    public function writeDataToPausedShouldReturnFalse()
+    public function writeDataToPausedShouldReturnFalse(): void
     {
         $through = new ThroughStream();
         $through->pause();
@@ -250,7 +250,7 @@ public function writeDataToPausedShouldReturnFalse()
     }
 
     /** @test */
-    public function writeDataToResumedShouldReturnTrue()
+    public function writeDataToResumedShouldReturnTrue(): void
     {
         $through = new ThroughStream();
         $through->pause();
@@ -260,21 +260,21 @@ public function writeDataToResumedShouldReturnTrue()
     }
 
     /** @test */
-    public function itShouldBeReadableByDefault()
+    public function itShouldBeReadableByDefault(): void
     {
         $through = new ThroughStream();
         $this->assertTrue($through->isReadable());
     }
 
     /** @test */
-    public function itShouldBeWritableByDefault()
+    public function itShouldBeWritableByDefault(): void
     {
         $through = new ThroughStream();
         $this->assertTrue($through->isWritable());
     }
 
     /** @test */
-    public function closeShouldCloseOnce()
+    public function closeShouldCloseOnce(): void
     {
         $through = new ThroughStream();
 
@@ -287,7 +287,7 @@ public function closeShouldCloseOnce()
     }
 
     /** @test */
-    public function doubleCloseShouldCloseOnce()
+    public function doubleCloseShouldCloseOnce(): void
     {
         $through = new ThroughStream();
 
@@ -301,7 +301,7 @@ public function doubleCloseShouldCloseOnce()
     }
 
     /** @test */
-    public function pipeShouldPipeCorrectly()
+    public function pipeShouldPipeCorrectly(): void
     {
         $output = $this->createMock(WritableStreamInterface::class);
         $output->expects($this->any())->method('isWritable')->willReturn(True);
diff --git a/tests/UtilTest.php b/tests/UtilTest.php
index cd00c8b..6bac04c 100644
--- a/tests/UtilTest.php
+++ b/tests/UtilTest.php
@@ -15,7 +15,7 @@
  */
 class UtilTest extends TestCase
 {
-    public function testPipeReturnsDestinationStream()
+    public function testPipeReturnsDestinationStream(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         assert($readable instanceof ReadableStreamInterface);
@@ -28,7 +28,7 @@ public function testPipeReturnsDestinationStream()
         $this->assertSame($writable, $ret);
     }
 
-    public function testPipeNonReadableSourceShouldDoNothing()
+    public function testPipeNonReadableSourceShouldDoNothing(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -49,7 +49,7 @@ public function testPipeNonReadableSourceShouldDoNothing()
         Util::pipe($readable, $writable);
     }
 
-    public function testPipeIntoNonWritableDestinationShouldPauseSource()
+    public function testPipeIntoNonWritableDestinationShouldPauseSource(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -74,7 +74,7 @@ public function testPipeIntoNonWritableDestinationShouldPauseSource()
         Util::pipe($readable, $writable);
     }
 
-    public function testPipeClosingDestPausesSource()
+    public function testPipeClosingDestPausesSource(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable
@@ -93,7 +93,7 @@ public function testPipeClosingDestPausesSource()
         $writable->close();
     }
 
-    public function testPipeWithEnd()
+    public function testPipeWithEnd(): void
     {
         $readable = new Stub\ReadableStreamStub();
 
@@ -112,7 +112,7 @@ public function testPipeWithEnd()
         $readable->end();
     }
 
-    public function testPipeWithoutEnd()
+    public function testPipeWithoutEnd(): void
     {
         $readable = new Stub\ReadableStreamStub();
 
@@ -131,7 +131,7 @@ public function testPipeWithoutEnd()
         $readable->end();
     }
 
-    public function testPipeWithTooSlowWritableShouldPauseReadable()
+    public function testPipeWithTooSlowWritableShouldPauseReadable(): void
     {
         $readable = new Stub\ReadableStreamStub();
 
@@ -154,7 +154,7 @@ public function testPipeWithTooSlowWritableShouldPauseReadable()
         $this->assertTrue($readable->paused);
     }
 
-    public function testPipeWithTooSlowWritableShouldResumeOnDrain()
+    public function testPipeWithTooSlowWritableShouldResumeOnDrain(): void
     {
         $readable = new Stub\ReadableStreamStub();
 
@@ -184,11 +184,13 @@ public function testPipeWithTooSlowWritableShouldResumeOnDrain()
         $this->assertFalse($readable->paused);
     }
 
-    public function testPipeWithWritableResourceStream()
+    public function testPipeWithWritableResourceStream(): void
     {
         $readable = new Stub\ReadableStreamStub();
 
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createMock(LoopInterface::class);
         assert($loop instanceof LoopInterface);
         $buffer = new WritableResourceStream($stream, $loop);
@@ -203,7 +205,7 @@ public function testPipeWithWritableResourceStream()
         $this->assertSame('hello, I am some random data', stream_get_contents($stream));
     }
 
-    public function testPipeSetsUpListeners()
+    public function testPipeSetsUpListeners(): void
     {
         $source = new ThroughStream();
         $dest = new ThroughStream();
@@ -219,7 +221,7 @@ public function testPipeSetsUpListeners()
         $this->assertCount(1, $dest->listeners('drain'));
     }
 
-    public function testPipeClosingSourceRemovesListeners()
+    public function testPipeClosingSourceRemovesListeners(): void
     {
         $source = new ThroughStream();
         $dest = new ThroughStream();
@@ -233,7 +235,7 @@ public function testPipeClosingSourceRemovesListeners()
         $this->assertCount(0, $dest->listeners('drain'));
     }
 
-    public function testPipeClosingDestRemovesListeners()
+    public function testPipeClosingDestRemovesListeners(): void
     {
         $source = new ThroughStream();
         $dest = new ThroughStream();
@@ -247,7 +249,7 @@ public function testPipeClosingDestRemovesListeners()
         $this->assertCount(0, $dest->listeners('drain'));
     }
 
-    public function testPipeDuplexIntoSelfEndsOnEnd()
+    public function testPipeDuplexIntoSelfEndsOnEnd(): void
     {
         $readable = $this->createMock(ReadableStreamInterface::class);
         $readable->expects($this->any())->method('isReadable')->willReturn(true);
@@ -267,7 +269,7 @@ public function testPipeDuplexIntoSelfEndsOnEnd()
     }
 
     /** @test */
-    public function forwardEventsShouldSetupForwards()
+    public function forwardEventsShouldSetupForwards(): void
     {
         $source = new ThroughStream();
         $target = new ThroughStream();
diff --git a/tests/WritableResourceStreamTest.php b/tests/WritableResourceStreamTest.php
index df3e815..bc4fc42 100644
--- a/tests/WritableResourceStreamTest.php
+++ b/tests/WritableResourceStreamTest.php
@@ -2,6 +2,7 @@
 
 namespace React\Tests\Stream;
 
+use PHPUnit\Framework\MockObject\MockObject;
 use React\EventLoop\LoopInterface;
 use React\Stream\WritableResourceStream;
 use function Clue\StreamFilter\append as filter_append;
@@ -12,17 +13,20 @@ class WritableResourceStreamTest extends TestCase
      * @covers React\Stream\WritableResourceStream::__construct
      * @doesNotPerformAssertions
      */
-    public function testConstructor()
+    public function testConstructor(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         new WritableResourceStream($stream, $loop);
     }
 
-    public function testConstructWithoutLoopAssignsLoopAutomatically()
+    public function testConstructWithoutLoopAssignsLoopAutomatically(): void
     {
         $resource = fopen('php://temp', 'r+');
+        assert(is_resource($resource));
 
         $stream = new WritableResourceStream($resource);
 
@@ -37,11 +41,12 @@ public function testConstructWithoutLoopAssignsLoopAutomatically()
      * @covers React\Stream\WritableResourceStream::__construct
      * @doesNotPerformAssertions
      */
-    public function testConstructorWithExcessiveMode()
+    public function testConstructorWithExcessiveMode(): void
     {
         // excessive flags are ignored for temp streams, so we have to use a file stream
-        $name = tempnam(sys_get_temp_dir(), 'test');
-        $stream = @fopen($name, 'w+eANYTHING');
+        $name = (string) tempnam(sys_get_temp_dir(), 'test');
+        $stream = fopen($name, 'w+eANYTHING');
+        assert(is_resource($stream));
         unlink($name);
 
         $loop = $this->createLoopMock();
@@ -52,7 +57,7 @@ public function testConstructorWithExcessiveMode()
     /**
      * @covers React\Stream\WritableResourceStream::__construct
      */
-    public function testConstructorThrowsIfNotAValidStreamResource()
+    public function testConstructorThrowsIfNotAValidStreamResource(): void
     {
         $stream = null;
         $loop = $this->createLoopMock();
@@ -64,9 +69,11 @@ public function testConstructorThrowsIfNotAValidStreamResource()
     /**
      * @covers React\Stream\WritableResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionOnReadOnlyStream()
+    public function testConstructorThrowsExceptionOnReadOnlyStream(): void
     {
         $stream = fopen('php://temp', 'r');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $this->expectException(\InvalidArgumentException::class);
@@ -76,11 +83,12 @@ public function testConstructorThrowsExceptionOnReadOnlyStream()
     /**
      * @covers React\Stream\WritableResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionOnReadOnlyStreamWithExcessiveMode()
+    public function testConstructorThrowsExceptionOnReadOnlyStreamWithExcessiveMode(): void
     {
         // excessive flags are ignored for temp streams, so we have to use a file stream
-        $name = tempnam(sys_get_temp_dir(), 'test');
+        $name = (string) tempnam(sys_get_temp_dir(), 'test');
         $stream = fopen($name, 'reANYTHING');
+        assert(is_resource($stream));
         unlink($name);
 
         $loop = $this->createLoopMock();
@@ -91,13 +99,15 @@ public function testConstructorThrowsExceptionOnReadOnlyStreamWithExcessiveMode(
     /**
      * @covers React\Stream\WritableResourceStream::__construct
      */
-    public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlocking()
+    public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlocking(): void
     {
         if (!in_array('blocking', stream_get_wrappers())) {
             stream_wrapper_register('blocking', 'React\Tests\Stream\EnforceBlockingWrapper');
         }
 
         $stream = fopen('blocking://test', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $this->expectException(\RuntimeException::class);
@@ -108,9 +118,11 @@ public function testConstructorThrowsExceptionIfStreamDoesNotSupportNonBlocking(
      * @covers React\Stream\WritableResourceStream::write
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testWrite()
+    public function testWrite(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createWriteableLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop);
@@ -124,9 +136,11 @@ public function testWrite()
     /**
      * @covers React\Stream\WritableResourceStream::write
      */
-    public function testWriteWithDataDoesAddResourceToLoop()
+    public function testWriteWithDataDoesAddResourceToLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('addWriteStream')->with($this->equalTo($stream));
 
@@ -139,9 +153,11 @@ public function testWriteWithDataDoesAddResourceToLoop()
      * @covers React\Stream\WritableResourceStream::write
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testEmptyWriteDoesNotAddToLoop()
+    public function testEmptyWriteDoesNotAddToLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->never())->method('addWriteStream');
 
@@ -155,9 +171,10 @@ public function testEmptyWriteDoesNotAddToLoop()
      * @covers React\Stream\WritableResourceStream::write
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testWriteReturnsFalseWhenWritableResourceStreamIsFull()
+    public function testWriteReturnsFalseWhenWritableResourceStreamIsFull(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
 
         $preventWrites = true;
         $loop = $this->createLoopMock();
@@ -182,9 +199,11 @@ public function testWriteReturnsFalseWhenWritableResourceStreamIsFull()
     /**
      * @covers React\Stream\WritableResourceStream::write
      */
-    public function testWriteReturnsFalseWhenWritableResourceStreamIsExactlyFull()
+    public function testWriteReturnsFalseWhenWritableResourceStreamIsExactlyFull(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop, 3);
@@ -196,9 +215,11 @@ public function testWriteReturnsFalseWhenWritableResourceStreamIsExactlyFull()
      * @covers React\Stream\WritableResourceStream::write
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testWriteDetectsWhenOtherSideIsClosed()
+    public function testWriteDetectsWhenOtherSideIsClosed(): void
     {
-        list($a, $b) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        assert(is_array($pair));
+        [$a, $b] = $pair;
 
         $loop = $this->createWriteableLoopMock();
 
@@ -214,9 +235,11 @@ public function testWriteDetectsWhenOtherSideIsClosed()
      * @covers React\Stream\WritableResourceStream::write
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testEmitsDrainAfterWriteWhichExceedsBuffer()
+    public function testEmitsDrainAfterWriteWhichExceedsBuffer(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop, 2);
@@ -231,9 +254,11 @@ public function testEmitsDrainAfterWriteWhichExceedsBuffer()
      * @covers React\Stream\WritableResourceStream::write
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testWriteInDrain()
+    public function testWriteInDrain(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop, 2);
@@ -255,9 +280,11 @@ public function testWriteInDrain()
      * @covers React\Stream\WritableResourceStream::write
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testDrainAfterWrite()
+    public function testDrainAfterWrite(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop, 2);
@@ -271,9 +298,11 @@ public function testDrainAfterWrite()
     /**
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testDrainAfterWriteWillRemoveResourceFromLoopWithoutClosing()
+    public function testDrainAfterWriteWillRemoveResourceFromLoopWithoutClosing(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('removeWriteStream')->with($stream);
 
@@ -290,9 +319,11 @@ public function testDrainAfterWriteWillRemoveResourceFromLoopWithoutClosing()
     /**
      * @covers React\Stream\WritableResourceStream::handleWrite
      */
-    public function testClosingDuringDrainAfterWriteWillRemoveResourceFromLoopOnceAndClose()
+    public function testClosingDuringDrainAfterWriteWillRemoveResourceFromLoopOnceAndClose(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $loop->expects($this->once())->method('removeWriteStream')->with($stream);
 
@@ -311,9 +342,11 @@ public function testClosingDuringDrainAfterWriteWillRemoveResourceFromLoopOnceAn
     /**
      * @covers React\Stream\WritableResourceStream::end
      */
-    public function testEndWithoutDataClosesImmediatelyIfWritableResourceStreamIsEmpty()
+    public function testEndWithoutDataClosesImmediatelyIfWritableResourceStreamIsEmpty(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop);
@@ -328,9 +361,11 @@ public function testEndWithoutDataClosesImmediatelyIfWritableResourceStreamIsEmp
     /**
      * @covers React\Stream\WritableResourceStream::end
      */
-    public function testEndWithoutDataDoesNotCloseIfWritableResourceStreamIsFull()
+    public function testEndWithoutDataDoesNotCloseIfWritableResourceStreamIsFull(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop);
@@ -347,9 +382,11 @@ public function testEndWithoutDataDoesNotCloseIfWritableResourceStreamIsFull()
     /**
      * @covers React\Stream\WritableResourceStream::end
      */
-    public function testEndWithDataClosesImmediatelyIfWritableResourceStreamFlushes()
+    public function testEndWithDataClosesImmediatelyIfWritableResourceStreamFlushes(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $filterBuffer = '';
         $loop = $this->createLoopMock();
 
@@ -373,9 +410,11 @@ public function testEndWithDataClosesImmediatelyIfWritableResourceStreamFlushes(
     /**
      * @covers React\Stream\WritableResourceStream::end
      */
-    public function testEndWithDataDoesNotCloseImmediatelyIfWritableResourceStreamIsFull()
+    public function testEndWithDataDoesNotCloseImmediatelyIfWritableResourceStreamIsFull(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop);
@@ -396,9 +435,11 @@ public function testEndWithDataDoesNotCloseImmediatelyIfWritableResourceStreamIs
      * @covers React\Stream\WritableResourceStream::isWritable
      * @covers React\Stream\WritableResourceStream::close
      */
-    public function testClose()
+    public function testClose(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop);
@@ -415,9 +456,11 @@ public function testClose()
     /**
      * @covers React\Stream\WritableResourceStream::close
      */
-    public function testClosingAfterWriteRemovesStreamFromLoop()
+    public function testClosingAfterWriteRemovesStreamFromLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $buffer = new WritableResourceStream($stream, $loop);
 
@@ -430,9 +473,11 @@ public function testClosingAfterWriteRemovesStreamFromLoop()
     /**
      * @covers React\Stream\WritableResourceStream::close
      */
-    public function testClosingWithoutWritingDoesNotRemoveStreamFromLoop()
+    public function testClosingWithoutWritingDoesNotRemoveStreamFromLoop(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
         $buffer = new WritableResourceStream($stream, $loop);
 
@@ -444,9 +489,11 @@ public function testClosingWithoutWritingDoesNotRemoveStreamFromLoop()
     /**
      * @covers React\Stream\WritableResourceStream::close
      */
-    public function testDoubleCloseWillEmitOnlyOnce()
+    public function testDoubleCloseWillEmitOnlyOnce(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $loop = $this->createLoopMock();
 
         $buffer = new WritableResourceStream($stream, $loop);
@@ -460,9 +507,11 @@ public function testDoubleCloseWillEmitOnlyOnce()
      * @covers React\Stream\WritableResourceStream::write
      * @covers React\Stream\WritableResourceStream::close
      */
-    public function testWritingToClosedWritableResourceStreamShouldNotWriteToStream()
+    public function testWritingToClosedWritableResourceStreamShouldNotWriteToStream(): void
     {
         $stream = fopen('php://temp', 'r+');
+        assert(is_resource($stream));
+
         $filterBuffer = '';
         $loop = $this->createLoopMock();
 
@@ -481,13 +530,16 @@ public function testWritingToClosedWritableResourceStreamShouldNotWriteToStream(
         $this->assertSame('', $filterBuffer);
     }
 
-    public function testWritingToClosedStream()
+    public function testWritingToClosedStream(): void
     {
         if ('Darwin' === PHP_OS) {
             $this->markTestSkipped('OS X issue with shutting down pair for writing');
         }
 
-        list($a, $b) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        assert(is_array($pair));
+        [$a, $b] = $pair;
+
         $loop = $this->createLoopMock();
 
         $error = null;
@@ -508,7 +560,7 @@ public function testWritingToClosedStream()
         $this->assertEqualsIgnoringCase('Unable to write to stream: fwrite(): send of 3 bytes failed with errno=32 Broken pipe', $error->getMessage());
     }
 
-    private function createWriteableLoopMock()
+    private function createWriteableLoopMock(): LoopInterface
     {
         $loop = $this->createLoopMock();
         $loop
@@ -521,8 +573,10 @@ private function createWriteableLoopMock()
         return $loop;
     }
 
-    private function createLoopMock()
+    /** @return MockObject&LoopInterface */
+    private function createLoopMock(): MockObject
     {
+        /** @var MockObject&LoopInterface */
         return $this->createMock(LoopInterface::class);
     }
 }