Add SseDecoder to decode Server-Sent Events from any readable stream

README.md

@@ -25,6 +25,7 @@ of [ReactPHP](https://reactphp.org/)'s event-driven architecture.
         * [MessageEvent::$data](#messageeventdata)
         * [MessageEvent::$lastEventId](#messageeventlasteventid)
         * [MessageEvent::$type](#messageeventtype)
+    * [SseDecoder](#ssedecoder)
 * [Install](#install)
 * [Tests](#tests)
 * [License](#license)
@@ -325,6 +326,146 @@ See also [`message` event](#message-event).
 If the message does not contain a `event` field or the `event` field is empty,
 the `$type` property will default to `message`.
 
+### SseDecoder
+
+The `SseDecoder` class is responsible for decoding the Server-Sent Events (SSE)
+wire protocol from any readable byte stream.
+
+Unlike the [`EventSource`](#eventsource) class which implements the higher-level
+HTML5 EventSource API (HTTP `GET` request, automatic reconnection, `readyState`
+and `Last-Event-ID` handling), this class only decodes the `text/event-stream`
+wire protocol. This makes it reusable whenever you already have a readable
+stream of SSE data, such as a streaming HTTP response to a custom request. This
+is common for LLM streaming APIs that return SSE over an HTTP `POST` request:
+
+```php
+$browser = new React\Http\Browser();
+
+$response = await($browser->requestStreaming(
+    'POST',
+    'https://api.example.com/v1/messages',
+    $headers,
+    $body
+));
+assert($response instanceof Psr\Http\Message\ResponseInterface);
+
+$stream = $response->getBody();
+assert($stream instanceof React\Stream\ReadableStreamInterface);
+
+$sse = new Clue\React\EventSource\SseDecoder($stream);
+$sse->on('data', function (Clue\React\EventSource\MessageEvent $message) {
+    $data = json_decode($message->data);
+
+    if ($data?->type === 'content_block_delta') {
+        echo $data->delta->text;
+    }
+});
+```
+
+Its constructor requires a readable stream that emits the raw SSE bytes:
+
+```php
+$sse = new Clue\React\EventSource\SseDecoder($stream);
+```
+
+When resuming a previously interrupted stream, you can pass the last event ID to
+continue from, so any message without an `id` field inherits it:
+
+```php
+$sse = new Clue\React\EventSource\SseDecoder($stream, $lastEventId);
+```
+
+The `SseDecoder` implements ReactPHP's
+[`ReadableStreamInterface`](https://github.com/reactphp/stream#readablestreaminterface)
+and emits a `data` event with a [`MessageEvent` object](#messageevent) for each
+incoming event. It is commonly used for transporting structured data such as JSON:
+
+```
+data: {"name":"Alice","age":30}
+
+data: {"name":"Bob","age":50}
+```
+```php
+$sse->on('data', function (Clue\React\EventSource\MessageEvent $message) {
+    $data = json_decode($message->data);
+    echo "{$data->name} is {$data->age} years old" . PHP_EOL;
+});
+```
+
+ReactPHP's underlying streams emit chunks of data strings and make no assumption
+about their lengths. These chunks do not necessarily represent complete SSE
+messages, as a single message may be broken up into multiple data chunks.
+This class reassembles these messages by buffering incomplete ones.
+
+Each event is dispatched as soon as its terminating blank line has been read.
+As per the SSE specification, an event with an empty data buffer (such as a
+comment or a lone `retry` field) will not be dispatched, and an incomplete event
+left in the buffer when the input stream ends will be discarded.
+
+The `id` and `retry` fields will be saved in the `string $lastEventId = ''`
+and `?float $lastRetryTime = null` properties respectively. They are
+assigned to the decoder before each `data` event is emitted as they may appear
+on events that are never dispatched. You can read their value from any event:
+
+```php
+$sse->on('close', function () use ($sse) {
+    printf(
+        'Reconnect in %.1f seconds with last event ID %s' . PHP_EOL,
+        $sse->lastRetryTime ?? 3.0,
+        $sse->lastEventId
+    );
+});
+```
+
+If the underlying stream emits an `error` event, it will forward this `error`
+event and then `close` the input stream. As per the SSE specs, any unknown or
+malformed fields will be ignored and will not cause an `error` event:
+
+```php
+$sse->on('error', function (Exception $error) {
+    // an error occurred, stream will close next
+});
+```
+
+If the underlying stream emits an `end` event, it will clear any incomplete
+data from the buffer and emit a final `end` event:
+
+```php
+$sse->on('end', function () {
+    // stream successfully ended, stream will close next
+});
+```
+
+If either the underlying stream or the `SseDecoder` is closed, it will forward
+the `close` event:
+
+```php
+$sse->on('close', function () {
+    // stream closed
+    // possibly after an "end" event or due to an "error" event
+});
+```
+
+The `close(): void` method can be used to explicitly close the `SseDecoder`
+and its underlying stream:
+
+```php
+$sse->close();
+```
+
+The `pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface`
+method can be used to forward all data to the given destination stream.
+Please note that the `SseDecoder` emits structured `MessageEvent` objects,
+while many writable streams expect only data chunks:
+
+```php
+$map = new ThroughStream(fn (MessageEvent $message) => $message->data);
+$sse->pipe($map)->pipe($logger);
+```
+
+For more details, see ReactPHP's
+[`ReadableStreamInterface`](https://github.com/reactphp/stream#readablestreaminterface).
+
 ## Install
 
 The recommended way to install this library is [through Composer](https://getcomposer.org/).

composer.json

@@ -15,7 +15,8 @@
         "evenement/evenement": "^3.0 || ^2.0",
         "react/event-loop": "^1.6",
         "react/http": "^1.11",
-        "react/promise": "^3.3 || ^2.10 || ^1.2.1"
+        "react/promise": "^3.3 || ^2.10 || ^1.2.1",
+        "react/stream": "^1.4"
     },
     "require-dev": {
         "phpunit/phpunit": "^9.6 || ^8.5 || ^5.7 || ^4.8.36"

src/EventSource.php

@@ -217,29 +217,19 @@ private function request()
             $stream = $response->getBody();
             assert($stream instanceof ReadableStreamInterface);
 
-            $buffer = '';
-            $stream->on('data', function ($chunk) use (&$buffer, $stream) {
-                $messageEvents = preg_split(
-                    '/(?:\r\n|\r(?!\n)|\n){2}/S',
-                    $buffer . $chunk
-                );
-                $buffer = array_pop($messageEvents);
-
-                foreach ($messageEvents as $data) {
-                    $message = MessageEvent::parse($data, $this->lastEventId, $this->reconnectTime);
-                    $this->lastEventId = $message->lastEventId;
-
-                    if ($message->data !== '') {
-                        $this->emit($message->type, array($message));
-                        if ($this->readyState === self::CLOSED) {
-                            break;
-                        }
-                    }
-                }
+            // resume from the last event ID received, it persists across reconnects
+            $sse = new SseDecoder($stream, $this->lastEventId);
+            $sse->on('data', function (MessageEvent $message) {
+                $this->emit($message->type, array($message));
             });
 
-            $stream->on('close', function () use (&$buffer) {
-                $buffer = '';
+            $sse->on('close', function () use ($sse) {
+                // `id` and `retry` may arrive on events that are never dispatched
+                $this->lastEventId = $sse->lastEventId;
+                if ($sse->lastRetryTime !== null) {
+                    $this->reconnectTime = $sse->lastRetryTime;
+                }
+
                 $this->request = null;
                 if ($this->readyState === self::OPEN) {
                     $this->readyState = self::CONNECTING;

src/SseDecoder.php

@@ -0,0 +1,149 @@
+<?php
+
+namespace Clue\React\EventSource;
+
+use Evenement\EventEmitter;
+use React\Stream\ReadableStreamInterface;
+use React\Stream\Util;
+use React\Stream\WritableStreamInterface;
+
+/**
+ * The `SseDecoder` reads from a plain byte stream and emits a `MessageEvent` for each Server-Sent Event (SSE).
+ *
+ * Unlike the `EventSource` class which implements the higher-level HTML5
+ * EventSource API (HTTP `GET` request, automatic reconnection, `readyState`
+ * and `Last-Event-ID` handling), this class only decodes the SSE wire protocol.
+ * This makes it reusable for any readable stream of `text/event-stream` data,
+ * such as a streaming HTTP response to a custom request (think LLM streaming
+ * APIs using an HTTP `POST` request).
+ */
+class SseDecoder extends EventEmitter implements ReadableStreamInterface
+{
+    /** @var ReadableStreamInterface */
+    private $input;
+
+    /** @var string */
+    private $buffer = '';
+
+    /** @var bool */
+    private $closed = false;
+
+    /**
+     * @var string (read-only) last event ID received from the `id` field or an empty string if not given
+     * @psalm-readonly-allow-private-mutation
+     */
+    public $lastEventId = '';
+
+    /**
+     * @var ?float (read-only) reconnection time in seconds from the `retry` field or null if not given
+     * @psalm-readonly-allow-private-mutation
+     */
+    public $lastRetryTime = null;
+
+    /**
+     * @param ReadableStreamInterface $input       readable byte stream emitting `text/event-stream` data
+     * @param string                  $lastEventId optional last event ID to resume from a previous stream
+     */
+    public function __construct(ReadableStreamInterface $input, $lastEventId = '')
+    {
+        $this->input = $input;
+        $this->lastEventId = $lastEventId;
+
+        if (!$input->isReadable()) {
+            $this->close();
+            return;
+        }
+
+        $this->input->on('data', array($this, 'handleData'));
+        $this->input->on('end', array($this, 'handleEnd'));
+        $this->input->on('error', array($this, 'handleError'));
+        $this->input->on('close', array($this, 'close'));
+    }
+
+    public function isReadable()
+    {
+        return !$this->closed;
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        $this->closed = true;
+        $this->buffer = '';
+
+        $this->input->close();
+
+        $this->emit('close');
+        $this->removeAllListeners();
+    }
+
+    public function pause()
+    {
+        $this->input->pause();
+    }
+
+    public function resume()
+    {
+        $this->input->resume();
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
+    /** @internal */
+    public function handleData($data)
+    {
+        if (!\is_string($data)) {
+            $this->handleError(new \UnexpectedValueException('Expected stream to emit string, but got ' . \gettype($data)));
+            return;
+        }
+
+        $this->buffer .= $data;
+
+        // keep parsing while a complete event (terminated by a blank line) has been found
+        while (\preg_match('/(?:\r\n|\r(?!\n)|\n){2}/S', $this->buffer, $match, \PREG_OFFSET_CAPTURE) === 1) {
+            // read event up until blank line and remove from buffer (including trailing blank line)
+            $event = (string) \substr($this->buffer, 0, $match[0][1]);
+            $this->buffer = (string) \substr($this->buffer, $match[0][1] + \strlen($match[0][0]));
+
+            // `id` and `retry` configure the event stream rather than a single message and may
+            // arrive on events that are never dispatched, so they are assigned to this stream
+            // before the `data` event is emitted and can be read at any time
+            $message = MessageEvent::parse($event, $this->lastEventId, $this->lastRetryTime);
+            $this->lastEventId = $message->lastEventId;
+
+            // dispatch event unless its data buffer is empty (as per SSE specs)
+            if ($message->data !== '') {
+                $this->emit('data', array($message));
+
+                if ($this->closed) {
+                    return;
+                }
+            }
+        }
+    }
+
+    /** @internal */
+    public function handleEnd()
+    {
+        // discard any incomplete event left in the buffer (as per SSE specs)
+        if (!$this->closed) {
+            $this->emit('end');
+            $this->close();
+        }
+    }
+
+    /** @internal */
+    public function handleError(\Exception $error)
+    {
+        $this->emit('error', array($error));
+        $this->close();
+    }
+}