feat(router): support server-sent events (#1260)

Co-authored-by: Brent Roose <[email protected]>

Author: innocenzi

packages/http/src/ContentType.php

@@ -27,6 +27,7 @@ enum ContentType: string
     case DOCX = 'application/vnd.openxmlformats-officedocument.wordprocessingml.document';
     case EOT = 'application/vnd.ms-fontobject';
     case EPUB = 'application/epub+zip';
+    case EVENT_STREAM = 'text/event-stream';
     case GZ = 'application/gzip';
     case GIF = 'image/gif';
     case HTML = 'text/html';

packages/http/src/Responses/EventStream.php

@@ -0,0 +1,40 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tempest\Http\Responses;
+
+use Closure;
+use Generator;
+use Tempest\DateTime\Duration;
+use Tempest\Http\ContentType;
+use Tempest\Http\IsResponse;
+use Tempest\Http\Response;
+use Tempest\Http\Status;
+
+final class EventStream implements Response
+{
+    use IsResponse;
+
+    public Duration $sleep;
+
+    public function __construct(
+        Closure $callback,
+        int|Duration $sleep = 1000,
+        Status $status = Status::OK,
+    ) {
+        $this->setContentType(ContentType::EVENT_STREAM);
+        $this->addHeader('X-Accel-Buffering', 'no');
+        $this->addHeader('Connection', 'keep-alive');
+        $this->addHeader('Cache-Control', 'no-cache');
+
+        $this->status = $status;
+        $this->body = $this->createGeneratorFromCallback($callback);
+        $this->sleep = is_int($sleep) ? Duration::milliseconds($sleep) : $sleep;
+    }
+
+    public function createGeneratorFromCallback($callback): Generator
+    {
+        yield from $callback();
+    }
+}

packages/http/src/ServerSentEvent.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace Tempest\Http;
+
+/**
+ * Represents a message streamed through server-sent events.
+ */
+final class ServerSentEvent
+{
+    public function __construct(
+        public mixed $data,
+        public string $event = 'message',
+    ) {}
+}

packages/router/src/GenericResponseSender.php

@@ -9,7 +9,10 @@
 use Tempest\Http\Header;
 use Tempest\Http\Response;
 use Tempest\Http\Responses\Download;
+use Tempest\Http\Responses\EventStream;
 use Tempest\Http\Responses\File;
+use Tempest\Http\ServerSentEvent;
+use Tempest\Support\Json;
 use Tempest\View\View;
 use Tempest\View\ViewRenderer;
 
@@ -22,10 +25,8 @@ public function __construct(
     public function send(Response $response): Response
     {
         ob_start();
-
         $this->sendHeaders($response);
         ob_flush();
-
         $this->sendContent($response);
         ob_end_flush();
 
@@ -68,6 +69,11 @@ private function resolveHeaders(Response $response): Generator
 
     private function sendContent(Response $response): void
     {
+        if ($response instanceof EventStream) {
+            $this->sendEventStream($response);
+            return;
+        }
+
         $body = $response->body;
 
         if ($response instanceof File || $response instanceof Download) {
@@ -82,4 +88,35 @@ private function sendContent(Response $response): void
 
         ob_flush();
     }
+
+    private function sendEventStream(EventStream $response): void
+    {
+        if (ob_get_level() > 0) {
+            ob_end_flush();
+        }
+
+        foreach ($response->body as $message) {
+            if (connection_aborted()) {
+                break;
+            }
+
+            $event = 'message';
+            $data = Json\encode($message);
+
+            if ($message instanceof ServerSentEvent) {
+                $event = $message->event;
+                $data = Json\encode($message->data);
+            }
+
+            echo "event: {$event}\n";
+            echo "data: {$data}";
+            echo "\n\n";
+
+            if (ob_get_level() > 0) {
+                ob_flush();
+            }
+
+            flush();
+        }
+    }
 }

packages/router/src/ResponseSenderInitializer.php

@@ -4,6 +4,7 @@
 
 namespace Tempest\Router;
 
+use Tempest\Clock\Clock;
 use Tempest\Container\Container;
 use Tempest\Container\Initializer;
 use Tempest\Container\Singleton;

tests/Integration/Http/GenericResponseSenderTest.php

@@ -6,8 +6,10 @@
 
 use Tempest\Http\GenericResponse;
 use Tempest\Http\Responses\Download;
+use Tempest\Http\Responses\EventStream;
 use Tempest\Http\Responses\File;
 use Tempest\Http\Responses\Ok;
+use Tempest\Http\ServerSentEvent;
 use Tempest\Http\Status;
 use Tempest\Router\GenericResponseSender;
 use Tests\Tempest\Integration\FrameworkIntegrationTestCase;
@@ -110,4 +112,39 @@ public function test_view_body(): void
 
         $this->assertStringContainsString('Hello Brent!', $output);
     }
+
+    public function test_stream(): void
+    {
+        ob_start();
+        $response = new EventStream(fn () => yield 'hello');
+        $responseSender = $this->container->get(GenericResponseSender::class);
+        $responseSender->send($response);
+        $output = ob_get_clean();
+
+        // restore phpunit's output buffer
+        ob_start();
+
+        $this->assertStringContainsString('event: message', $output);
+        $this->assertStringContainsString('data: "hello"', $output);
+    }
+
+    public function test_stream_with_custom_event(): void
+    {
+        ob_start();
+        $response = new EventStream(function () {
+            yield new ServerSentEvent(data: 'hello', event: 'first');
+            yield new ServerSentEvent(data: 'goodbye', event: 'last');
+        });
+        $responseSender = $this->container->get(GenericResponseSender::class);
+        $responseSender->send($response);
+        $output = ob_get_clean();
+
+        // restore phpunit's output buffer
+        ob_start();
+
+        $this->assertStringContainsString('event: first', $output);
+        $this->assertStringContainsString('data: "hello"', $output);
+        $this->assertStringContainsString('event: last', $output);
+        $this->assertStringContainsString('data: "goodbye"', $output);
+    }
 }