diff --git a/src/AmphpIpcPeer.php b/src/AmphpIpcPeer.php index c057ba2..54cee5d 100644 --- a/src/AmphpIpcPeer.php +++ b/src/AmphpIpcPeer.php @@ -6,8 +6,8 @@ use Amp\ByteStream\ReadableResourceStream; use Amp\ByteStream\WritableResourceStream; use Amp\DeferredFuture; -use StreamIpc\Transport\AmpByteStreamMessageTransport; use Revolt\EventLoop; +use StreamIpc\Transport\AmpByteStreamMessageTransport; use function Amp\Future\awaitFirst; use function is_resource; diff --git a/src/Envelope/ResponsePromise.php b/src/Envelope/ResponsePromise.php index b8638bb..da72a04 100644 --- a/src/Envelope/ResponsePromise.php +++ b/src/Envelope/ResponsePromise.php @@ -16,6 +16,7 @@ final class ResponsePromise { public const DEFAULT_TIMEOUT = 30.0; + /** @var ?Message The response message, null if not yet received. */ private ?Message $response = null; diff --git a/src/IpcSession.php b/src/IpcSession.php index baf9e2f..5c5c11e 100644 --- a/src/IpcSession.php +++ b/src/IpcSession.php @@ -7,7 +7,7 @@ use StreamIpc\Envelope\RequestEnvelope; use StreamIpc\Envelope\ResponseEnvelope; use StreamIpc\Envelope\ResponsePromise; -use StreamIpc\Message\LogMessage; +use StreamIpc\Message\ErrorMessage; use StreamIpc\Message\Message; use StreamIpc\Transport\MessageTransport; use StreamIpc\Transport\StreamClosedException; @@ -91,7 +91,7 @@ public function dispatch(Message $envelope): void } catch (Throwable $e) { try { $this->transport->send( - new ResponseEnvelope($envelope->id, new LogMessage('Error in dispatch: ' . $e->getMessage(), 'error')) + new ResponseEnvelope($envelope->id, new ErrorMessage('Error in dispatch', $e)) ); } catch (StreamClosedException) { throw $e; diff --git a/src/Message/ErrorMessage.php b/src/Message/ErrorMessage.php new file mode 100644 index 0000000..31c7a09 --- /dev/null +++ b/src/Message/ErrorMessage.php @@ -0,0 +1,26 @@ +getMessage() . PHP_EOL . $exception->getTraceAsString(); + } + + $this->errorMessage = trim($errorMessage); + } + + public function toString(): string + { + return $this->errorMessage; + } +} diff --git a/src/Message/LogMessage.php b/src/Message/LogMessage.php index 11ae1ec..a3f4d14 100644 --- a/src/Message/LogMessage.php +++ b/src/Message/LogMessage.php @@ -12,6 +12,8 @@ */ final readonly class LogMessage implements Message { + public const LEVEL_JUNK = 'junk'; + /** * Constructs a new LogMessage. * diff --git a/src/NativeIpcPeer.php b/src/NativeIpcPeer.php index 1666dac..4852e44 100644 --- a/src/NativeIpcPeer.php +++ b/src/NativeIpcPeer.php @@ -3,9 +3,9 @@ namespace StreamIpc; +use RuntimeException; use StreamIpc\Transport\MessageTransport; use StreamIpc\Transport\NativeMessageTransport; -use RuntimeException; /** * IPC peer implementation that works with standard PHP stream resources. @@ -142,6 +142,7 @@ public function tick(?float $timeout = null): void $usec = (int)(($timeout - $sec) * 1e6); } + $writes = $except = null; if (@stream_select($reads, $writes, $except, $sec, $usec) <= 0) { // no streams ready or error occurred return; diff --git a/src/Serialization/JsonMessageSerializer.php b/src/Serialization/JsonMessageSerializer.php index fbe16ab..86b302e 100644 --- a/src/Serialization/JsonMessageSerializer.php +++ b/src/Serialization/JsonMessageSerializer.php @@ -5,9 +5,9 @@ namespace StreamIpc\Serialization; use JsonException; -use StreamIpc\Message\LogMessage; -use StreamIpc\Message\Message; use ReflectionClass; +use StreamIpc\Message\ErrorMessage; +use StreamIpc\Message\Message; use Throwable; use function class_exists; @@ -100,17 +100,17 @@ public function deserialize(string $data): Message { try { $decoded = json_decode($data, true, 512, JSON_THROW_ON_ERROR); - } catch (Throwable $e) { - return new LogMessage($data, 'error'); + } catch (JsonException $e) { + return new ErrorMessage('json_decode failed:' . $data, $e); } if (!is_array($decoded) || !isset($decoded['__class'])) { - return new LogMessage($data, 'error'); + return new ErrorMessage('Invalid data to unserialize: ' . $data); } $class = $decoded['__class']; if (!class_exists($class)) { - return new LogMessage("Unknown class: {$class}", 'error'); + return new ErrorMessage("Unknown class: {$class}"); } try { @@ -127,11 +127,11 @@ public function deserialize(string $data): Message $prop->setValue($inst, $this->fromArray($value)); } } catch (Throwable $e) { - return new LogMessage($data, 'error'); + return new ErrorMessage('Failed to unserialize data:' . $data, $e); } if (!$inst instanceof Message) { - return new LogMessage("Decoded object is not a Message: {$data}", 'error'); + return new ErrorMessage("Decoded object is not a Message: {$data}"); } return $inst; diff --git a/src/Transport/FrameCodec.php b/src/Transport/FrameCodec.php index d0a6570..8dc926d 100644 --- a/src/Transport/FrameCodec.php +++ b/src/Transport/FrameCodec.php @@ -4,6 +4,7 @@ namespace StreamIpc\Transport; use RuntimeException; +use StreamIpc\Message\ErrorMessage; use StreamIpc\Message\LogMessage; use StreamIpc\Message\Message; use StreamIpc\Serialization\MessageSerializer; @@ -11,17 +12,19 @@ final class FrameCodec { - public const MAGIC = "\xF3\x4A\x9D\xE2"; + public const MAGIC = "\xF3\x4A\x9D\xE2"; private const MAGIC_LEN = 4; - private const LEN_LEN = 4; + private const LEN_LEN = 4; private string $buffer = ''; - private int $scanPos = 0; + + private int $scanPos = 0; public function __construct( private readonly MessageSerializer $serializer, private readonly ?int $maxFrame = null - ) {} + ) { + } public function pack(Message $message): string { @@ -48,7 +51,7 @@ public function feed(string $chunk): array $junkLen = strlen($this->buffer) - $overlap; if ($junkLen > 0) { - $messages[] = new LogMessage(substr($this->buffer, 0, $junkLen), 'error'); + $messages[] = new LogMessage(substr($this->buffer, 0, $junkLen), LogMessage::LEVEL_JUNK); $this->buffer = substr($this->buffer, $junkLen); // keep only overlap } $this->scanPos = 0; @@ -57,7 +60,7 @@ public function feed(string $chunk): array // ── 2. junk before header ────────────────────────────────────────────── if ($pos > 0) { - $messages[] = new LogMessage(substr($this->buffer, 0, $pos), 'error'); + $messages[] = new LogMessage(substr($this->buffer, 0, $pos), LogMessage::LEVEL_JUNK); $this->buffer = substr($this->buffer, $pos); $this->scanPos = 0; } @@ -69,10 +72,10 @@ public function feed(string $chunk): array // ── 4. parse 32-bit BE length fast ───────────────────────────────────── $lenOffset = self::MAGIC_LEN; - $length = (ord($this->buffer[$lenOffset ]) << 24) + $length = (ord($this->buffer[$lenOffset ]) << 24) | (ord($this->buffer[$lenOffset + 1]) << 16) - | (ord($this->buffer[$lenOffset + 2]) << 8) - | ord($this->buffer[$lenOffset + 3]); + | (ord($this->buffer[$lenOffset + 2]) << 8) + | ord($this->buffer[$lenOffset + 3]); if ($this->maxFrame !== null && $length > $this->maxFrame) { throw new RuntimeException("Frame length $length exceeds max {$this->maxFrame}"); @@ -84,16 +87,18 @@ public function feed(string $chunk): array } // ── 5. extract payload & consume buffer ──────────────────────────────── - $payload = substr($this->buffer, + $payload = substr( + $this->buffer, self::MAGIC_LEN + self::LEN_LEN, - $length); + $length + ); $this->buffer = substr($this->buffer, $frameSize); $this->scanPos = 0; try { $messages[] = $this->serializer->deserialize($payload); } catch (Throwable $e) { - $messages[] = new LogMessage($payload, 'error'); + $messages[] = new ErrorMessage('Failed to deserialize message payload', $e); } } diff --git a/src/Transport/NativeFrameReader.php b/src/Transport/NativeFrameReader.php index 57b0f6f..41bf00d 100644 --- a/src/Transport/NativeFrameReader.php +++ b/src/Transport/NativeFrameReader.php @@ -26,6 +26,7 @@ final class NativeFrameReader * @param $serializer MessageSerializer used for incoming frames. * @param $maxFrame ?int Maximum allowed size for a single message frame. */ + /** * @param resource $stream */ diff --git a/tests/Unit/FrameCodecTest.php b/tests/Unit/FrameCodecTest.php index 406ba26..522e956 100644 --- a/tests/Unit/FrameCodecTest.php +++ b/tests/Unit/FrameCodecTest.php @@ -8,6 +8,7 @@ use StreamIpc\Serialization\NativeMessageSerializer; use StreamIpc\Tests\Fixtures\SimpleMessage; use StreamIpc\Message\LogMessage; +use StreamIpc\Message\ErrorMessage; use RuntimeException; use StreamIpc\Message\Message; use Throwable; @@ -183,7 +184,7 @@ public function deserialize(string $data): Message $msgs = $codec->feed($frame); self::assertCount(1, $msgs); - self::assertInstanceOf(LogMessage::class, $msgs[0]); + self::assertInstanceOf(ErrorMessage::class, $msgs[0]); } /** 8. Feeding an empty string must be a no-op ******************************/ diff --git a/tests/Unit/IpcSessionTest.php b/tests/Unit/IpcSessionTest.php index 506e354..ba1180c 100644 --- a/tests/Unit/IpcSessionTest.php +++ b/tests/Unit/IpcSessionTest.php @@ -5,6 +5,7 @@ use StreamIpc\Tests\Fixtures\FakeTransport; use StreamIpc\Tests\Fixtures\SimpleMessage; use StreamIpc\Message\LogMessage; +use StreamIpc\Message\ErrorMessage; use StreamIpc\Envelope\RequestEnvelope; use StreamIpc\Envelope\ResponseEnvelope; use PHPUnit\Framework\TestCase; @@ -58,7 +59,7 @@ public function testDispatchResponseStoresMessage(): void unset($promise); } - public function testDispatchErrorFromHandlerSendsLogMessage(): void + public function testDispatchErrorFromHandlerSendsErrorMessage(): void { $peer = new TestPeer(); $transport = new FakeTransport(); @@ -73,7 +74,8 @@ public function testDispatchErrorFromHandlerSendsLogMessage(): void $this->assertCount(1, $transport->sent); $this->assertInstanceOf(ResponseEnvelope::class, $transport->sent[0]); $this->assertSame('123', $transport->sent[0]->id); - $this->assertStringContainsString('boom', $transport->sent[0]->response->message); + $this->assertInstanceOf(ErrorMessage::class, $transport->sent[0]->response); + $this->assertStringContainsString('boom', $transport->sent[0]->response->toString()); } public function testRequestCreatesPromiseAndSendsEnvelope(): void diff --git a/tests/Unit/JsonMessageSerializerTest.php b/tests/Unit/JsonMessageSerializerTest.php index 585b71b..364af4e 100644 --- a/tests/Unit/JsonMessageSerializerTest.php +++ b/tests/Unit/JsonMessageSerializerTest.php @@ -2,7 +2,7 @@ namespace StreamIpc\Tests\Unit; use StreamIpc\Serialization\JsonMessageSerializer; -use StreamIpc\Message\LogMessage; +use StreamIpc\Message\ErrorMessage; use StreamIpc\Tests\Fixtures\SimpleMessage; use StreamIpc\Tests\Fixtures\ComplexMessage; use PHPUnit\Framework\TestCase; @@ -23,22 +23,22 @@ public function testRoundTripWithNestedObjects(): void $this->assertSame([1, 2], $decoded->list); } - public function testUnknownClassReturnsLogMessage(): void + public function testUnknownClassReturnsErrorMessage(): void { $serializer = new JsonMessageSerializer(); $json = json_encode(['__class' => 'NoSuchClass', 'foo' => 'bar']); $decoded = $serializer->deserialize($json); - $this->assertInstanceOf(LogMessage::class, $decoded); - $this->assertSame('error', $decoded->level); + $this->assertInstanceOf(ErrorMessage::class, $decoded); + $this->assertStringContainsString('Unknown class', $decoded->toString()); } - public function testInvalidJsonProducesLogMessage(): void + public function testInvalidJsonProducesErrorMessage(): void { $serializer = new JsonMessageSerializer(); $decoded = $serializer->deserialize('{invalid json'); - $this->assertInstanceOf(LogMessage::class, $decoded); - $this->assertSame('{invalid json', $decoded->message); + $this->assertInstanceOf(ErrorMessage::class, $decoded); + $this->assertStringContainsString('json_decode failed', $decoded->toString()); } public function testDeserializedObjectNotImplementingMessage(): void @@ -46,7 +46,7 @@ public function testDeserializedObjectNotImplementingMessage(): void $serializer = new JsonMessageSerializer(); $objData = json_encode(['__class' => \stdClass::class]); $decoded = $serializer->deserialize($objData); - $this->assertInstanceOf(LogMessage::class, $decoded); - $this->assertSame('error', $decoded->level); + $this->assertInstanceOf(ErrorMessage::class, $decoded); + $this->assertStringContainsString('Decoded object is not a Message', $decoded->toString()); } }