diff --git a/src/Responses/StreamResponse.php b/src/Responses/StreamResponse.php index cddda7aa..d4ce383a 100644 --- a/src/Responses/StreamResponse.php +++ b/src/Responses/StreamResponse.php @@ -17,6 +17,10 @@ */ final class StreamResponse implements ResponseHasMetaInformationContract, ResponseStreamContract { + private const STREAM_READ_SIZE = 64 * 1024; + + private string $lineBuffer = ''; + /** * Creates a new Stream Response instance. * @@ -34,64 +38,133 @@ public function __construct( */ public function getIterator(): Generator { - while (! $this->response->getBody()->eof()) { - $line = $this->readLine($this->response->getBody()); + $body = $this->response->getBody(); + $event = null; - $event = null; + while (($line = $this->readLine($body)) !== null) { if (str_starts_with($line, 'event:')) { $event = trim(substr($line, strlen('event:'))); - $line = $this->readLine($this->response->getBody()); + + unset($line); + + continue; } if (! str_starts_with($line, 'data:')) { + $event = null; + + unset($line); + continue; } - $data = trim(substr($line, strlen('data:'))); + $data = substr($line, strlen('data:')); + + unset($line); + + if (strlen($data) <= 16 && trim($data) === '[DONE]') { + unset($data); - if ($data === '[DONE]') { break; } - /** @var array{error?: array{message: string|array, type: string, code: string}, type?: string} $response */ - $response = json_decode($data, true, flags: JSON_THROW_ON_ERROR); + /** @var array{error?: array{message: string|array, type: string, code: string}, type?: string} $attributes */ + $attributes = json_decode($data, true, flags: JSON_THROW_ON_ERROR); + + unset($data); - if (isset($response['error'])) { - throw new ErrorException($response['error'], $this->response); + if (isset($attributes['error'])) { + throw new ErrorException($attributes['error'], $this->response); } $skippableTypes = ['ping', 'keepalive', 'response.keep_alive']; - if (isset($response['type']) && in_array($response['type'], $skippableTypes, true)) { + + if (isset($attributes['type']) && in_array($attributes['type'], $skippableTypes, true)) { + $event = null; + + unset($attributes); + continue; } if ($event !== null) { - $response['__event'] = $event; + $attributes['__event'] = $event; } - $response['__meta'] = $this->meta(); - yield $this->responseClass::from($response); + $attributes['__meta'] = $this->meta(); + + $streamEvent = $this->responseClass::from($attributes); + + $event = null; + + unset($attributes); + + yield $streamEvent; + + unset($streamEvent); } } /** * Read a line from the stream. */ - private function readLine(StreamInterface $stream): string + private function readLine(StreamInterface $stream): ?string { - $buffer = ''; + $newLinePosition = strpos($this->lineBuffer, "\n"); + + if ($newLinePosition !== false) { + $lineLength = $newLinePosition + 1; + + if ($lineLength === strlen($this->lineBuffer)) { + $line = $this->lineBuffer; + $this->lineBuffer = ''; + + return $line; + } + + $line = substr($this->lineBuffer, 0, $lineLength); + $this->lineBuffer = substr($this->lineBuffer, $lineLength); + + return $line; + } while (! $stream->eof()) { - if ('' === ($byte = $stream->read(1))) { - return $buffer; + $chunk = $stream->read(self::STREAM_READ_SIZE); + + if ($chunk === '') { + continue; } - $buffer .= $byte; - if ($byte === "\n") { - break; + + // Split the fresh chunk before appending it to avoid copying a large buffered line. + $newLinePosition = strpos($chunk, "\n"); + + if ($newLinePosition === false) { + $this->lineBuffer .= $chunk; + + continue; + } + + $lineLength = $newLinePosition + 1; + $line = $this->lineBuffer; + $this->lineBuffer = substr($chunk, $lineLength); + + if ($lineLength === strlen($chunk)) { + $line .= $chunk; + } else { + $line .= substr($chunk, 0, $lineLength); } + + return $line; } - return $buffer; + if ($this->lineBuffer === '') { + return null; + } + + $line = $this->lineBuffer; + $this->lineBuffer = ''; + + return $line; } public function meta(): MetaInformation diff --git a/tests/Responses/StreamResponse.php b/tests/Responses/StreamResponse.php new file mode 100644 index 00000000..71b4323b --- /dev/null +++ b/tests/Responses/StreamResponse.php @@ -0,0 +1,122 @@ +readCalls++; + $this->largestRead = max($this->largestRead, $length); + + if ($this->readCalls === $this->emptyReadCall) { + return ''; + } + + if ($this->maximumBytesPerRead !== null) { + $length = min($length, $this->maximumBytesPerRead); + } + + return $this->stream->read($length); + } +} + +test('reads large SSE events in chunks without losing buffered events', function () { + $largeEncryptedContent = str_repeat('a', 128 * 1024); + + $events = [ + [ + 'type' => 'response.output_item.done', + 'output_index' => 0, + 'sequence_number' => 1, + 'item' => [ + 'id' => 'cmp_large', + 'encrypted_content' => $largeEncryptedContent, + 'type' => 'compaction', + 'created_by' => 'user', + ], + ], + [ + 'type' => 'response.output_item.done', + 'output_index' => 1, + 'sequence_number' => 2, + 'item' => [ + 'id' => 'cmp_buffered', + 'encrypted_content' => 'buffered content', + 'type' => 'compaction', + 'created_by' => 'user', + ], + ], + ]; + + $body = implode('', array_map( + fn (array $event): string => "event: response.output_item.done\n". + 'data: '.json_encode($event, flags: JSON_THROW_ON_ERROR)."\n\n", + $events, + )).'data: [DONE]'; + + $stream = new StreamResponseReadTrackingStream(Utils::streamFor($body)); + $response = new Response(body: $stream); + $streamResponse = new StreamResponse(CreateStreamedResponse::class, $response); + + $result = iterator_to_array($streamResponse); + + expect($result) + ->toHaveCount(2) + ->and($result[0]->response->item) + ->toBeInstanceOf(OutputCompaction::class) + ->encryptedContent->toBe($largeEncryptedContent) + ->and($result[1]->response->item) + ->toBeInstanceOf(OutputCompaction::class) + ->encryptedContent->toBe('buffered content') + ->and($stream->largestRead)->toBe(64 * 1024) + ->and($stream->readCalls)->toBeLessThan(10); +}); + +test('retries an empty read before EOF without losing the buffered line', function () { + $attributes = [ + 'type' => 'response.output_item.done', + 'output_index' => 0, + 'sequence_number' => 1, + 'item' => [ + 'id' => 'cmp_after_empty_read', + 'encrypted_content' => 'complete content', + 'type' => 'compaction', + 'created_by' => 'user', + ], + ]; + $body = 'data: '.json_encode($attributes, flags: JSON_THROW_ON_ERROR)."\n"; + + $stream = new StreamResponseReadTrackingStream(Utils::streamFor($body)); + $stream->maximumBytesPerRead = 10; + $stream->emptyReadCall = 2; + + $response = new Response(body: $stream); + $streamResponse = new StreamResponse(CreateStreamedResponse::class, $response); + + $result = iterator_to_array($streamResponse); + + expect($result) + ->toHaveCount(1) + ->and($result[0]->response->item) + ->toBeInstanceOf(OutputCompaction::class) + ->encryptedContent->toBe('complete content'); +});