From 0aa200ac4999a6992a13a09c3b642ddb3c2e2b87 Mon Sep 17 00:00:00 2001
From: riki137 <richard@popelis.sk>
Date: Tue, 24 Jun 2025 14:07:41 +0200
Subject: [PATCH] add InvalidStreamException and enhance error handling in
 NativeIpcPeer

---
 src/InvalidStreamException.php                | 27 +++++++++++++++++++
 src/NativeIpcPeer.php                         | 23 +++++++++++++---
 tests/Unit/NativeIpcPeerInvalidStreamTest.php | 26 ++++++++++++++++++
 3 files changed, 73 insertions(+), 3 deletions(-)
 create mode 100644 src/InvalidStreamException.php
 create mode 100644 tests/Unit/NativeIpcPeerInvalidStreamTest.php

diff --git a/src/InvalidStreamException.php b/src/InvalidStreamException.php
new file mode 100644
index 0000000..0aa70dd
--- /dev/null
+++ b/src/InvalidStreamException.php
@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace StreamIpc;
+
+use RuntimeException;
+use Throwable;
+
+/**
+ * Exception thrown when a registered stream becomes invalid.
+ */
+class InvalidStreamException extends RuntimeException
+{
+    private IpcSession $session;
+
+    public function __construct(IpcSession $session, ?string $message = null, int $code = 0, ?Throwable $previous = null)
+    {
+        $this->session = $session;
+        parent::__construct($message ?? 'Invalid stream resource', $code, $previous);
+    }
+
+    public function getSession(): IpcSession
+    {
+        return $this->session;
+    }
+}
diff --git a/src/NativeIpcPeer.php b/src/NativeIpcPeer.php
index aa100d0..074b2b8 100644
--- a/src/NativeIpcPeer.php
+++ b/src/NativeIpcPeer.php
@@ -4,6 +4,8 @@
 namespace StreamIpc;
 
 use RuntimeException;
+use TypeError;
+use ValueError;
 use StreamIpc\Transport\MessageTransport;
 use StreamIpc\Transport\NativeMessageTransport;
 
@@ -145,9 +147,24 @@ public function tick(?float $timeout = null): void
         }
 
         $writes = $except = null;
-        if (@stream_select($reads, $writes, $except, $sec, $usec) <= 0) {
-            // no streams ready or error occurred
-            return;
+        try {
+            if (@stream_select($reads, $writes, $except, $sec, $usec) <= 0) {
+                // no streams ready or error occurred
+                return;
+            }
+        } catch (TypeError|ValueError $e) {
+            foreach ($this->readSet as $key => $stream) {
+                $test = [$stream];
+                $w = $ex = null;
+                try {
+                    stream_select($test, $w, $ex, 0, 0);
+                } catch (TypeError|ValueError) {
+                    [$session] = $this->fdMap[$key];
+                    throw new InvalidStreamException($session, null, 0, $e);
+                }
+            }
+
+            throw $e;
         }
 
         foreach ($reads as $stream) {
diff --git a/tests/Unit/NativeIpcPeerInvalidStreamTest.php b/tests/Unit/NativeIpcPeerInvalidStreamTest.php
new file mode 100644
index 0000000..035a140
--- /dev/null
+++ b/tests/Unit/NativeIpcPeerInvalidStreamTest.php
@@ -0,0 +1,26 @@
+<?php
+namespace StreamIpc\Tests\Unit;
+
+use PHPUnit\Framework\TestCase;
+use StreamIpc\InvalidStreamException;
+use StreamIpc\NativeIpcPeer;
+
+final class NativeIpcPeerInvalidStreamTest extends TestCase
+{
+    public function testTickThrowsInvalidStreamExceptionWithSession(): void
+    {
+        [$a, $b] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        $peer = new NativeIpcPeer();
+        $session = $peer->createStreamSession($a, $a, $b);
+        fclose($b);
+
+        try {
+            $peer->tick();
+            $this->fail('No exception thrown');
+        } catch (InvalidStreamException $e) {
+            $this->assertSame($session, $e->getSession());
+        } finally {
+            fclose($a);
+        }
+    }
+}