factory = new Psr17Factory(); } #[TestDox('SSE stream is aborted before the buffer can exceed the configured cap')] public function testSseBufferIsBoundedByConfiguredCap(): void { $transport = $this->createTransport(maxSseBufferBytes: 64); $state = new ClientState(); $state->addPendingRequest(1, 30); $transport->setState($state); // A server that streams data without ever sending the "\n\n" delimiter. $this->setActiveStream($transport, $this->factory->createStream(str_repeat('a', 4096))); $this->invokeProcessSseStream($transport); self::assertSame('', $this->readPrivate($transport, 'sseBuffer'), 'buffer must be cleared on abort'); self::assertNull($this->readPrivate($transport, 'activeStream'), 'stream must be released on abort'); } #[TestDox('aborting the SSE stream fails the in-flight request immediately instead of waiting for its timeout')] public function testAbortFailsPendingRequestFast(): void { $transport = $this->createTransport(maxSseBufferBytes: 64); $state = new ClientState(); $state->addPendingRequest(1, 30); $transport->setState($state); $this->setActiveStream($transport, $this->factory->createStream(str_repeat('a', 4096))); $this->invokeProcessSseStream($transport); $response = $state->consumeResponse(1); self::assertInstanceOf(Error::class, $response); self::assertSame(Error::INTERNAL_ERROR, $response->code); self::assertSame(1, $response->id); } #[TestDox('well-formed delimited events are parsed and dispatched')] public function testWellFormedEventsStillParse(): void { $transport = $this->createTransport(); $messages = []; $transport->onMessage(static function (string $message) use (&$messages): void { $messages[] = $message; }); $this->setActiveStream($transport, $this->factory->createStream("data: hello\n\ndata: world\n\n")); $this->invokeProcessSseStream($transport); self::assertSame(['hello', 'world'], $messages); } #[TestDox('the buffer cap must be a positive number of bytes')] public function testRejectsNonPositiveCap(): void { $this->expectException(InvalidArgumentException::class); $this->createTransport(maxSseBufferBytes: 0); } private function createTransport(int $maxSseBufferBytes = 8 * 1024 * 1024): HttpTransport { return new HttpTransport( endpoint: 'https://example.test/mcp', httpClient: $this->createMock(ClientInterface::class), requestFactory: $this->factory, streamFactory: $this->factory, maxSseBufferBytes: $maxSseBufferBytes, ); } private function setActiveStream(HttpTransport $transport, StreamInterface $stream): void { (new \ReflectionProperty($transport, 'activeStream'))->setValue($transport, $stream); } private function invokeProcessSseStream(HttpTransport $transport): void { (new \ReflectionMethod($transport, 'processSSEStream'))->invoke($transport); } private function readPrivate(HttpTransport $transport, string $property): mixed { return (new \ReflectionProperty($transport, $property))->getValue($transport); } }