symfony · fabpot · Aug 6, 2020 · May 4, 2020 · stof · Jun 12, 2020
diff --git a/src/Symfony/Component/HttpClient/CHANGELOG.md b/src/Symfony/Component/HttpClient/CHANGELOG.md
@@ -8,6 +8,7 @@ CHANGELOG
  * added support for pausing responses with a new `pause_handler` callable exposed as an info item
  * added `StreamableInterface` to ease turning responses into PHP streams
  * added `MockResponse::getRequestMethod()` and `getRequestUrl()` to allow inspecting which request has been sent
+ * added `EventSourceHttpClient` a Server-Sent events stream implementing the [EventSource specification](https://www.w3.org/TR/eventsource/#eventsource)
 
 5.1.0
 -----

diff --git a/src/Symfony/Component/HttpClient/Chunk/ServerSentEvent.php b/src/Symfony/Component/HttpClient/Chunk/ServerSentEvent.php
@@ -0,0 +1,79 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\HttpClient\Chunk;
+
+use Symfony\Contracts\HttpClient\ChunkInterface;
+
+/**
+ * @author Antoine Bluchet <soyuka@gmail.com>
+ * @author Nicolas Grekas <p@tchwork.com>
+ */
+final class ServerSentEvent extends DataChunk implements ChunkInterface
+{
+    private $data = '';
+    private $id = '';
+    private $type = 'message';
+    private $retry = 0;
+
+    public function __construct(string $content)
+    {
+        parent::__construct(-1, $content);
+
+        // remove BOM
+        if (0 === strpos($content, "\xEF\xBB\xBF")) {
+            $content = substr($content, 3);
+        }
+
+        foreach (preg_split("/(?:\r\n|[\r\n])/", $content) as $line) {
+            if (0 === $i = strpos($line, ':')) {
+                continue;
+            }
+
+            $i = false === $i ? \strlen($line) : $i;
+            $field = substr($line, 0, $i);
+            $i += 1 + (' ' === ($line[1 + $i] ?? ''));
+
+            switch ($field) {
+                case 'id': $this->id = substr($line, $i); break;
+                case 'event': $this->type = substr($line, $i); break;
+                case 'data': $this->data .= ('' === $this->data ? '' : "\n").substr($line, $i); break;
+                case 'retry':
+                    $retry = substr($line, $i);
+
+                    if ('' !== $retry && \strlen($retry) === strspn($retry, '0123456789')) {
+                        $this->retry = $retry / 1000.0;
+                    }
+                    break;
+            }
+        }
+    }
+
+    public function getId(): string
+    {
+        return $this->id;
+    }
+
+    public function getType(): string
+    {
+        return $this->type;
+    }
+
+    public function getData(): string
+    {
+        return $this->data;
+    }
+
+    public function getRetry(): float
+    {
+        return $this->retry;
+    }
+}
diff --git a/src/Symfony/Component/HttpClient/EventSourceHttpClient.php b/src/Symfony/Component/HttpClient/EventSourceHttpClient.php
@@ -0,0 +1,153 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\HttpClient;
+
+use Symfony\Component\HttpClient\Chunk\ServerSentEvent;
+use Symfony\Component\HttpClient\Exception\EventSourceException;
+use Symfony\Component\HttpClient\Response\AsyncContext;
+use Symfony\Component\HttpClient\Response\AsyncResponse;
+use Symfony\Contracts\HttpClient\ChunkInterface;
+use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
+use Symfony\Contracts\HttpClient\HttpClientInterface;
+use Symfony\Contracts\HttpClient\ResponseInterface;
+
+/**
+ * @author Antoine Bluchet <soyuka@gmail.com>
+ * @author Nicolas Grekas <p@tchwork.com>
+ */
+final class EventSourceHttpClient implements HttpClientInterface
+{
+    use AsyncDecoratorTrait;
+    use HttpClientTrait;
+
+    private $reconnectionTime;
+
+    public function __construct(HttpClientInterface $client = null, float $reconnectionTime = 10.0)
+    {
+        $this->client = $client ?? HttpClient::create();
+        $this->reconnectionTime = $reconnectionTime;
+    }
+
+    public function connect(string $url, array $options = []): ResponseInterface
+    {
+        return $this->request('GET', $url, self::mergeDefaultOptions($options, [
+            'buffer' => false,
+            'headers' => [
+                'Accept' => 'text/event-stream',
+                'Cache-Control' => 'no-cache',
+            ],
+        ], true));
+    }
+
+    public function request(string $method, string $url, array $options = []): ResponseInterface
+    {
+        $state = new class() {
+            public $buffer = null;
+            public $lastEventId = null;
+            public $reconnectionTime;
+            public $lastError = null;
+        };
+        $state->reconnectionTime = $this->reconnectionTime;
+
+        if ($accept = self::normalizeHeaders($options['headers'] ?? [])['accept'] ?? []) {
+            $state->buffer = \in_array($accept, [['Accept: text/event-stream'], ['accept: text/event-stream']], true) ? '' : null;
+        }
+
+        return new AsyncResponse($this->client, $method, $url, $options, static function (ChunkInterface $chunk, AsyncContext $context) use ($state, $method, $url, $options) {
+            if (null !== $state->buffer) {
+                $context->setInfo('reconnection_time', $state->reconnectionTime);
+                $isTimeout = false;
+            }
+            $lastError = $state->lastError;
+            $state->lastError = null;
+
+            try {
+                $isTimeout = $chunk->isTimeout();
+
+                if (null !== $chunk->getInformationalStatus()) {
+                    yield $chunk;
+
+                    return;
+                }
+            } catch (TransportExceptionInterface $e) {
+                $state->lastError = $lastError ?? microtime(true);
+
+                if (null === $state->buffer || ($isTimeout && microtime(true) - $state->lastError < $state->reconnectionTime)) {
+                    yield $chunk;
+                } else {
+                    $options['headers']['Last-Event-ID'] = $state->lastEventId;
+                    $state->buffer = '';
+                    $state->lastError = microtime(true);
+                    $context->getResponse()->cancel();
+                    $context->replaceRequest($method, $url, $options);
+                    if ($isTimeout) {
+                        yield $chunk;
+                    } else {
+                        $context->pause($state->reconnectionTime);
+                    }
+                }
+
+                return;
+            }
+
+            if ($chunk->isFirst()) {
+                if (preg_match('/^text\/event-stream(;|$)/i', $context->getHeaders()['content-type'][0] ?? '')) {
+                    $state->buffer = '';
+                } elseif (null !== $lastError || (null !== $state->buffer && 200 === $context->getStatusCode())) {
+                    throw new EventSourceException(sprintf('Response content-type is "%s" while "text/event-stream" was expected for "%s".', $context->getHeaders()['content-type'][0] ?? '', $context->getInfo('url')));
+                } else {
+                    $context->passthru();
+                }
+
+                if (null === $lastError) {
+                    yield $chunk;
+                }
+
+                return;
+            }
+
+            $rx = '/((?:\r\n|[\r\n]){2,})/';
+            $content = $state->buffer.$chunk->getContent();
+
+            if ($chunk->isLast()) {
+                $rx = substr_replace($rx, '|$', -2, 0);
+            }
+            $events = preg_split($rx, $content, -1, PREG_SPLIT_DELIM_CAPTURE);
+            $state->buffer = array_pop($events);
+
+            for ($i = 0; isset($events[$i]); $i += 2) {
+                $event = new ServerSentEvent($events[$i].$events[1 + $i]);
+
+                if ('' !== $event->getId()) {
+                    $context->setInfo('last_event_id', $state->lastEventId = $event->getId());
+                }
+
+                if ($event->getRetry()) {
+                    $context->setInfo('reconnection_time', $state->reconnectionTime = $event->getRetry());
+                }
+
+                yield $event;
+            }
+
+            if (preg_match('/^(?::[^\r\n]*+(?:\r\n|[\r\n]))+$/m', $state->buffer)) {
+                $content = $state->buffer;
+                $state->buffer = '';
+
+                yield $context->createChunk($content);
+            }
+
+            if ($chunk->isLast()) {
+                yield $chunk;
+            }
+        });
+    }
+}
diff --git a/src/Symfony/Component/HttpClient/Exception/EventSourceException.php b/src/Symfony/Component/HttpClient/Exception/EventSourceException.php
@@ -0,0 +1,21 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\HttpClient\Exception;
+
+use Symfony\Contracts\HttpClient\Exception\DecodingExceptionInterface;
+
+/**
+ * @author Nicolas Grekas <p@tchwork.com>
+ */
+final class EventSourceException extends \RuntimeException implements DecodingExceptionInterface
+{
+}
diff --git a/src/Symfony/Component/HttpClient/Tests/Chunk/ServerSentEventTest.php b/src/Symfony/Component/HttpClient/Tests/Chunk/ServerSentEventTest.php
@@ -0,0 +1,79 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\HttpClient\Tests\Chunk;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\HttpClient\Chunk\ServerSentEvent;
+
+/**
+ * @author Antoine Bluchet <soyuka@gmail.com>
+ */
+class ServerSentEventTest extends TestCase
+{
+    public function testParse()
+    {
+        $rawData = <<<STR
+data: test
+data:test
+id: 12
+event: testEvent
+
+STR;
+
+        $sse = new ServerSentEvent($rawData);
+        $this->assertSame("test\ntest", $sse->getData());
+        $this->assertSame('12', $sse->getId());
+        $this->assertSame('testEvent', $sse->getType());
+    }
+
+    public function testParseValid()
+    {
+        $rawData = <<<STR
+event: testEvent
+data
+
+STR;
+
+        $sse = new ServerSentEvent($rawData);
+        $this->assertSame('', $sse->getData());
+        $this->assertSame('', $sse->getId());
+        $this->assertSame('testEvent', $sse->getType());
+    }
+
+    public function testParseRetry()
+    {
+        $rawData = <<<STR
+retry: 12
+STR;
+        $sse = new ServerSentEvent($rawData);
+        $this->assertSame('', $sse->getData());
+        $this->assertSame('', $sse->getId());
+        $this->assertSame('message', $sse->getType());
+        $this->assertSame(0.012, $sse->getRetry());
+    }
+
+    public function testParseNewLine()
+    {
+        $rawData = <<<STR
+
+
+data: <tag>
+data
+data:   <foo />
+data:
+data: 
+data: </tag>
+STR;
+        $sse = new ServerSentEvent($rawData);
+        $this->assertSame("<tag>\n\n  <foo />\n\n\n</tag>", $sse->getData());
+    }
+}