ars-backend/app/Services/Agent/OpenAi/OpenAiStreamParser.php

<?php

namespace App\Services\Agent\OpenAi;

use Psr\Http\Message\StreamInterface;

class OpenAiStreamParser
{
    public function __construct(private readonly int $chunkSize = 1024)
    {
    }

    /**
     * Parses SSE data lines into payload strings.
     *
     * @return \Generator<int, string>
     */
    public function parse(StreamInterface $stream, ?callable $shouldStop = null): \Generator
    {
        $buffer = '';
        $eventData = '';

        while (! $stream->eof()) {
            if ($shouldStop && $shouldStop()) {
                break;
            }

            $chunk = $stream->read($this->chunkSize);

            if ($chunk === '') {
                usleep(10000);
                continue;
            }

            $buffer .= $chunk;

            while (($pos = strpos($buffer, "\n")) !== false) {
                $line = substr($buffer, 0, $pos);
                $buffer = substr($buffer, $pos + 1);
                $line = rtrim($line, "\r");

                if ($line === '') {
                    if ($eventData !== '') {
                        yield $eventData;
                        $eventData = '';
                    }
                    continue;
                }

                if (str_starts_with($line, 'data:')) {
                    $data = ltrim(substr($line, 5));
                    if ($eventData !== '') {
                        $eventData .= "\n";
                    }
                    $eventData .= $data;
                }
            }
        }

        if ($buffer !== '') {
            $line = rtrim($buffer, "\r");
            if (str_starts_with($line, 'data:')) {
                $data = ltrim(substr($line, 5));
                if ($eventData !== '') {
                    $eventData .= "\n";
                }
                $eventData .= $data;
            }
        }

        if ($eventData !== '') {
            yield $eventData;
        }
    }
}