diff --git a/README.md b/README.md
index 0f31aef..c48e390 100644
--- a/README.md
+++ b/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/).
diff --git a/composer.json b/composer.json
index 809fc08..083f44f 100644
--- a/composer.json
+++ b/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"
diff --git a/src/EventSource.php b/src/EventSource.php
index 72cbe51..2ebfd44 100644
--- a/src/EventSource.php
+++ b/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;
diff --git a/src/SseDecoder.php b/src/SseDecoder.php
new file mode 100644
index 0000000..7aa472b
--- /dev/null
+++ b/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();
+    }
+}
diff --git a/tests/EventSourceTest.php b/tests/EventSourceTest.php
index 3f7dc6d..02da9d3 100644
--- a/tests/EventSourceTest.php
+++ b/tests/EventSourceTest.php
@@ -657,6 +657,143 @@ public function testReconnectAfterStreamClosesUsesLastEventIdFromParsedEventStre
         $timerReconnect();
     }
 
+    public function testReconnectTwiceAfterSecondStreamWithoutIdStillUsesLastEventIdFromFirstStream()
+    {
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $timerReconnect = null;
+        $loop->expects($this->exactly(2))->method('addTimer')->with(
+            3.0,
+            $this->callback(function ($cb) use (&$timerReconnect) {
+                $timerReconnect = $cb;
+                return true;
+            })
+        );
+
+        $first = new Deferred();
+        $second = new Deferred();
+        $browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
+        $browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
+        $browser->expects($this->exactly(3))->method('requestStreaming')->withConsecutive(
+            ['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache']],
+            ['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache', 'Last-Event-ID' => '123']],
+            ['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache', 'Last-Event-ID' => '123']]
+        )->willReturnOnConsecutiveCalls(
+            $first->promise(),
+            $second->promise(),
+            new Promise(function () { })
+        );
+
+        $es = new EventSource('http://example.com', $browser, $loop);
+
+        $stream = new ThroughStream();
+        $first->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
+
+        $stream->write("id:123\n\n");
+        $stream->end();
+
+        $this->assertNotNull($timerReconnect);
+        $timerReconnect();
+
+        $stream = new ThroughStream();
+        $second->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
+
+        $stream->write("data:hello\n\n");
+        $stream->end();
+
+        $this->assertNotNull($timerReconnect);
+        $timerReconnect();
+    }
+
+    public function testReconnectTwiceAfterSecondStreamWithEmptyIdClearsLastEventIdForNextRequest()
+    {
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $timerReconnect = null;
+        $loop->expects($this->exactly(2))->method('addTimer')->with(
+            3.0,
+            $this->callback(function ($cb) use (&$timerReconnect) {
+                $timerReconnect = $cb;
+                return true;
+            })
+        );
+
+        $first = new Deferred();
+        $second = new Deferred();
+        $browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
+        $browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
+        $browser->expects($this->exactly(3))->method('requestStreaming')->withConsecutive(
+            ['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache']],
+            ['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache', 'Last-Event-ID' => '123']],
+            ['GET', 'http://example.com', ['Accept' => 'text/event-stream', 'Cache-Control' => 'no-cache']]
+        )->willReturnOnConsecutiveCalls(
+            $first->promise(),
+            $second->promise(),
+            new Promise(function () { })
+        );
+
+        $es = new EventSource('http://example.com', $browser, $loop);
+
+        $stream = new ThroughStream();
+        $first->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
+
+        $stream->write("id:123\n\n");
+        $stream->end();
+
+        $this->assertNotNull($timerReconnect);
+        $timerReconnect();
+
+        $stream = new ThroughStream();
+        $second->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
+
+        $stream->write("id:\ndata:hello\n\n");
+        $stream->end();
+
+        $this->assertNotNull($timerReconnect);
+        $timerReconnect();
+    }
+
+    public function testReconnectTwiceAfterSecondStreamWithoutRetryStillUsesRetryTimeFromFirstStream()
+    {
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $timerReconnect = null;
+        $loop->expects($this->exactly(2))->method('addTimer')->with(
+            2.543,
+            $this->callback(function ($cb) use (&$timerReconnect) {
+                $timerReconnect = $cb;
+                return true;
+            })
+        );
+
+        $first = new Deferred();
+        $second = new Deferred();
+        $browser = $this->getMockBuilder('React\Http\Browser')->disableOriginalConstructor()->getMock();
+        $browser->expects($this->once())->method('withRejectErrorResponse')->willReturnSelf();
+        $browser->expects($this->exactly(3))->method('requestStreaming')->willReturnOnConsecutiveCalls(
+            $first->promise(),
+            $second->promise(),
+            new Promise(function () { })
+        );
+
+        $es = new EventSource('http://example.com', $browser, $loop);
+
+        $stream = new ThroughStream();
+        $first->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
+
+        $stream->write("retry:2543\n\n");
+        $stream->end();
+
+        $this->assertNotNull($timerReconnect);
+        $timerReconnect();
+
+        $stream = new ThroughStream();
+        $second->resolve(new Response(200, array('Content-Type' => 'text/event-stream'), $stream));
+
+        $stream->write("data:hello\n\n");
+        $stream->end();
+
+        $this->assertNotNull($timerReconnect);
+        $timerReconnect();
+    }
+
     public function testReconnectAfterStreamClosesUsesSpecifiedRetryTime()
     {
         $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
diff --git a/tests/SseDecoderTest.php b/tests/SseDecoderTest.php
new file mode 100644
index 0000000..b229983
--- /dev/null
+++ b/tests/SseDecoderTest.php
@@ -0,0 +1,384 @@
+<?php
+
+namespace Clue\Tests\React\EventSource;
+
+use Clue\React\EventSource\MessageEvent;
+use Clue\React\EventSource\SseDecoder;
+use PHPUnit\Framework\TestCase;
+use React\Stream\ThroughStream;
+
+class SseDecoderTest extends TestCase
+{
+    /** @var ThroughStream */
+    private $input;
+
+    /** @var SseDecoder */
+    private $decoder;
+
+    /**
+     * @before
+     */
+    public function setUpDecoder()
+    {
+        $this->input = new ThroughStream();
+        $this->decoder = new SseDecoder($this->input);
+    }
+
+    public function testConstructWillAssignEmptyLastEventIdAndNullRetryTime()
+    {
+        $this->assertSame('', $this->decoder->lastEventId);
+        $this->assertNull($this->decoder->lastRetryTime);
+    }
+
+    public function testConstructWithLastEventIdWillUseAsIdForMessageWithoutId()
+    {
+        $this->decoder = new SseDecoder($this->input, '42');
+
+        $message = null;
+        $this->decoder->on('data', function ($m) use (&$message) {
+            $message = $m;
+        });
+
+        $this->input->emit('data', array("data: hello\n\n"));
+
+        $this->assertEquals('42', $message->lastEventId);
+        $this->assertSame('42', $this->decoder->lastEventId);
+    }
+
+    public function testEmitDataWillForwardMessageEventWithParsedData()
+    {
+        $message = null;
+        $this->decoder->on('data', function ($m) use (&$message) {
+            $message = $m;
+        });
+
+        $this->input->emit('data', array("data: hello\n\n"));
+
+        $this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
+        $this->assertEquals('hello', $message->data);
+        $this->assertEquals('message', $message->type);
+        $this->assertEquals('', $message->lastEventId);
+    }
+
+    public function testEmitDataWillForwardCustomEventTypeAndId()
+    {
+        $message = null;
+        $this->decoder->on('data', function ($m) use (&$message) {
+            $message = $m;
+        });
+
+        $this->input->emit('data', array("event: patch\nid: 1\ndata: hello\n\n"));
+
+        $this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
+        $this->assertEquals('hello', $message->data);
+        $this->assertEquals('patch', $message->type);
+        $this->assertEquals('1', $message->lastEventId);
+    }
+
+    public function testEmitDataWillForwardDataOverMultipleLinesCombined()
+    {
+        $message = null;
+        $this->decoder->on('data', function ($m) use (&$message) {
+            $message = $m;
+        });
+
+        $this->input->emit('data', array("data: hello\ndata: world\n\n"));
+
+        $this->assertEquals("hello\nworld", $message->data);
+    }
+
+    public function testEmitTwoEventsInSingleChunkWillForwardTwoMessageEvents()
+    {
+        $messages = [];
+        $this->decoder->on('data', function ($m) use (&$messages) {
+            $messages[] = $m;
+        });
+
+        $this->input->emit('data', array("data: hello\n\ndata: world\n\n"));
+
+        $this->assertCount(2, $messages);
+        $this->assertEquals('hello', $messages[0]->data);
+        $this->assertEquals('world', $messages[1]->data);
+    }
+
+    public function testEmitEventOverMultipleChunksWillForwardSingleMessageEventOnceComplete()
+    {
+        $messages = [];
+        $this->decoder->on('data', function ($m) use (&$messages) {
+            $messages[] = $m;
+        });
+
+        $this->input->emit('data', array("data: hel"));
+        $this->assertCount(0, $messages);
+
+        $this->input->emit('data', array("lo\n"));
+        $this->assertCount(0, $messages);
+
+        $this->input->emit('data', array("\n"));
+        $this->assertCount(1, $messages);
+        $this->assertEquals('hello', $messages[0]->data);
+    }
+
+    public function testEmitEventWillInheritLastEventIdAcrossEvents()
+    {
+        $messages = [];
+        $this->decoder->on('data', function ($m) use (&$messages) {
+            $messages[] = $m;
+        });
+
+        $this->input->emit('data', array("id: 100\ndata: first\n\ndata: second\n\n"));
+
+        $this->assertCount(2, $messages);
+        $this->assertEquals('100', $messages[0]->lastEventId);
+        $this->assertEquals('100', $messages[1]->lastEventId);
+    }
+
+    public function testEmitEventWithCarriageReturnLineFeedBoundaryWillForwardMessageEvent()
+    {
+        $message = null;
+        $this->decoder->on('data', function ($m) use (&$message) {
+            $message = $m;
+        });
+
+        $this->input->emit('data', array("data: hello\r\n\r\n"));
+
+        $this->assertInstanceOf('Clue\React\EventSource\MessageEvent', $message);
+        $this->assertEquals('hello', $message->data);
+    }
+
+    public function testEmitCommentOnlyEventWithoutDataWillNotForwardMessageEvent()
+    {
+        $this->decoder->on('data', function () {
+            $this->fail('Did not expect data event');
+        });
+
+        $this->input->emit('data', array(": this is a comment\n\n"));
+
+        $this->assertTrue($this->decoder->isReadable());
+    }
+
+    public function testEmitIncompleteEventWithoutBoundaryWillNotForwardMessageEvent()
+    {
+        $this->decoder->on('data', function () {
+            $this->fail('Did not expect data event');
+        });
+
+        $this->input->emit('data', array("data: hello\n"));
+
+        $this->assertTrue($this->decoder->isReadable());
+    }
+
+    public function testEmitIncompleteEventThenEndWillNotForwardMessageEventAndDiscardBuffer()
+    {
+        $this->decoder->on('data', function () {
+            $this->fail('Did not expect data event');
+        });
+        $ended = false;
+        $this->decoder->on('end', function () use (&$ended) {
+            $ended = true;
+        });
+
+        $this->input->emit('data', array("data: hello\n"));
+        $this->input->emit('end');
+
+        $this->assertTrue($ended);
+        $this->assertFalse($this->decoder->isReadable());
+    }
+
+    public function testEmitEndWillForwardEnd()
+    {
+        $ended = false;
+        $this->decoder->on('end', function () use (&$ended) {
+            $ended = true;
+        });
+        $this->decoder->on('close', function () use (&$ended) {
+            $this->assertTrue($ended);
+        });
+
+        $this->input->emit('end');
+
+        $this->assertTrue($ended);
+        $this->assertFalse($this->decoder->isReadable());
+    }
+
+    public function testEmitDataWithInvalidTypeWillForwardErrorAndClose()
+    {
+        $error = null;
+        $this->decoder->on('error', function ($e) use (&$error) {
+            $error = $e;
+        });
+        $this->decoder->on('close', function () use (&$error) {
+            $this->assertNotNull($error);
+        });
+
+        $this->input->emit('data', array(false));
+
+        $this->assertInstanceOf('UnexpectedValueException', $error);
+        $this->assertFalse($this->decoder->isReadable());
+    }
+
+    public function testEmitRetryOnlyEventWillAssignRetryTimeAndNotForwardMessageEvent()
+    {
+        $this->decoder->on('data', function () {
+            $this->fail('Did not expect data event');
+        });
+
+        $this->input->emit('data', array("retry: 2543\n\n"));
+
+        $this->assertSame(2.543, $this->decoder->lastRetryTime);
+    }
+
+    public function testEmitIdOnlyEventWillAssignLastEventIdAndNotForwardMessageEvent()
+    {
+        $this->decoder->on('data', function () {
+            $this->fail('Did not expect data event');
+        });
+
+        $this->input->emit('data', array("id: 42\n\n"));
+
+        $this->assertSame('42', $this->decoder->lastEventId);
+    }
+
+    public function testEmitEventWillAssignRetryTimeBeforeForwardingMessageEvent()
+    {
+        $retryTime = null;
+        $this->decoder->on('data', function () use (&$retryTime) {
+            $retryTime = $this->decoder->lastRetryTime;
+        });
+
+        $this->input->emit('data', array("retry: 2543\ndata: hello\n\n"));
+
+        $this->assertSame(2.543, $retryTime);
+    }
+
+    public function testEmitErrorEventWillForwardErrorAndClose()
+    {
+        $error = null;
+        $this->decoder->on('error', function ($e) use (&$error) {
+            $error = $e;
+        });
+        $this->decoder->on('close', function () use (&$error) {
+            $this->assertNotNull($error);
+        });
+
+        $exception = new \RuntimeException();
+        $this->input->emit('error', array($exception));
+
+        $this->assertSame($exception, $error);
+        $this->assertFalse($this->decoder->isReadable());
+    }
+
+    public function testClosingDecoderDuringDataEventWillNotForwardFurtherMessageEvents()
+    {
+        $messages = [];
+        $this->decoder->on('data', function ($m) use (&$messages) {
+            $messages[] = $m;
+        });
+        $this->decoder->on('data', array($this->decoder, 'close'));
+
+        $this->input->emit('data', array("data: hello\n\ndata: world\n\n"));
+
+        $this->assertCount(1, $messages);
+        $this->assertEquals('hello', $messages[0]->data);
+        $this->assertFalse($this->decoder->isReadable());
+    }
+
+    public function testClosingInputWillCloseDecoder()
+    {
+        $closed = false;
+        $this->decoder->on('close', function () use (&$closed) {
+            $closed = true;
+        });
+
+        $this->assertTrue($this->decoder->isReadable());
+
+        $this->input->close();
+
+        $this->assertTrue($closed);
+        $this->assertFalse($this->decoder->isReadable());
+    }
+
+    public function testClosingInputWillRemoveAllDataListeners()
+    {
+        $this->decoder->on('data', function () { });
+
+        $this->input->close();
+
+        $this->assertEquals([], $this->input->listeners('data'));
+        $this->assertEquals([], $this->decoder->listeners('data'));
+    }
+
+    public function testClosingDecoderWillCloseInput()
+    {
+        $closed = false;
+        $this->input->on('close', function () use (&$closed) {
+            $closed = true;
+        });
+
+        $this->assertTrue($this->decoder->isReadable());
+
+        $this->decoder->close();
+
+        $this->assertTrue($closed);
+        $this->assertFalse($this->decoder->isReadable());
+    }
+
+    public function testClosingDecoderTwiceWillCloseInputOnce()
+    {
+        $closed = 0;
+        $this->decoder->on('close', function () use (&$closed) {
+            ++$closed;
+        });
+
+        $this->decoder->close();
+        $this->decoder->close();
+
+        $this->assertEquals(1, $closed);
+    }
+
+    public function testUnreadableInputWillResultInUnreadableDecoder()
+    {
+        $this->input->close();
+        $this->decoder = new SseDecoder($this->input);
+
+        $this->assertFalse($this->decoder->isReadable());
+    }
+
+    public function testUnreadableInputWillNotAddAnyEventListeners()
+    {
+        $this->input->close();
+        $this->decoder = new SseDecoder($this->input);
+
+        $this->assertEquals([], $this->input->listeners('data'));
+        $this->assertEquals([], $this->decoder->listeners('data'));
+    }
+
+    public function testPipeReturnsDestStream()
+    {
+        $dest = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+
+        $ret = $this->decoder->pipe($dest);
+
+        $this->assertSame($dest, $ret);
+    }
+
+    public function testForwardPauseToInput()
+    {
+        $input = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $input->expects($this->once())->method('isReadable')->willReturn(true);
+        $input->expects($this->once())->method('pause');
+
+        $decoder = new SseDecoder($input);
+        $decoder->pause();
+    }
+
+    public function testForwardResumeToInput()
+    {
+        $input = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $input->expects($this->once())->method('isReadable')->willReturn(true);
+        $input->expects($this->once())->method('resume');
+
+        $decoder = new SseDecoder($input);
+        $decoder->resume();
+    }
+}