feat: 优化 SSE emitter 基础能力

src/Response/SseResponse.php

@@ -47,7 +47,7 @@ final class SseResponse extends Text implements StreamingResponseInterface
     private ?\Closure $onError;
 
     /**
-     * @param iterable<Event|string|array>|\Closure(EmitterInterface): void $producer
+     * @param iterable<Event|string|array|object>|\Closure(EmitterInterface): void $producer
      * @param array{
      *     heartbeat?: int,
      *     max_duration?: int,
@@ -115,8 +115,14 @@ public function send(): void
         @set_time_limit(0);
         @ignore_user_abort(true);
 
-        $emitter = new CgiEmitter($this->maxDuration);
+        $this->emit(new CgiEmitter($this->maxDuration));
+    }
 
+    /**
+     * 使用指定发送器执行生产者；便于测试和自定义运行时注入。
+     */
+    public function emit(EmitterInterface $emitter): void
+    {
         try {
             if ($this->retry !== null) {
                 $emitter->retry($this->retry);

src/Sse/AbstractEmitter.php

@@ -0,0 +1,59 @@
+<?php
+
+declare(strict_types=1);
+
+namespace FastD\Http\Sse;
+
+/**
+ * SSE 发送器公共实现。
+ *
+ * 统一帧归一化、注释帧、重连间隔和关闭状态；具体运行时只需实现 write()。
+ */
+abstract class AbstractEmitter implements EmitterInterface
+{
+    protected bool $closed = false;
+
+    public function send(string|array|object $event): void
+    {
+        if ($this->isClosed()) {
+            return;
+        }
+
+        $this->write($this->normalize($event)->format());
+    }
+
+    public function comment(string $text): void
+    {
+        if ($this->isClosed()) {
+            return;
+        }
+
+        $this->write(': ' . str_replace(["\r", "\n"], ' ', $text) . "\n\n");
+    }
+
+    public function retry(int $milliseconds): void
+    {
+        if ($milliseconds < 0 || $this->isClosed()) {
+            return;
+        }
+
+        $this->write('retry: ' . $milliseconds . "\n\n");
+    }
+
+    public function isClosed(): bool
+    {
+        return $this->closed;
+    }
+
+    public function close(): void
+    {
+        $this->closed = true;
+    }
+
+    abstract protected function write(string $frame): void;
+
+    protected function normalize(string|array|object $event): Event
+    {
+        return $event instanceof Event ? $event : Event::data($event);
+    }
+}

src/Sse/CgiEmitter.php

@@ -11,10 +11,8 @@
  * 是否停止。生产者在长时间阻塞（如等待上游）期间无法产生心跳——需要心跳
  * 的 AI 流式场景建议使用 Swoole 运行时。
  */
-final class CgiEmitter implements EmitterInterface
+final class CgiEmitter extends AbstractEmitter
 {
-    private bool $closed = false;
-
     /** @var \Closure(string): void */
     private \Closure $writer;
 
@@ -51,33 +49,6 @@ public function __construct(
         $this->deadline = $maxDuration > 0 ? ($this->clock)() + (float) $maxDuration : null;
     }
 
-    public function send(Event|string|array $event): void
-    {
-        if ($this->isClosed()) {
-            return;
-        }
-
-        $this->write($this->normalize($event)->format());
-    }
-
-    public function comment(string $text): void
-    {
-        if ($this->isClosed()) {
-            return;
-        }
-
-        $this->write(': ' . str_replace(["\r", "\n"], ' ', $text) . "\n\n");
-    }
-
-    public function retry(int $milliseconds): void
-    {
-        if ($this->isClosed() || $milliseconds < 0) {
-            return;
-        }
-
-        $this->write('retry: ' . $milliseconds . "\n\n");
-    }
-
     public function isClosed(): bool
     {
         if ($this->closed) {
@@ -99,18 +70,8 @@ public function isClosed(): bool
         return false;
     }
 
-    public function close(): void
-    {
-        $this->closed = true;
-    }
-
-    private function write(string $frame): void
+    protected function write(string $frame): void
     {
         ($this->writer)($frame);
     }
-
-    private function normalize(Event|string|array $event): Event
-    {
-        return $event instanceof Event ? $event : Event::data($event);
-    }
 }

src/Sse/EmitterInterface.php

@@ -15,9 +15,9 @@ interface EmitterInterface
     /**
      * 发送一条事件。
      *
-     * @param Event|string|array<array-key, mixed> $event string 作为 data；数组/对象 JSON 编码
+     * @param Event|string|array<array-key, mixed>|object $event string 作为 data；数组/对象 JSON 编码
      */
-    public function send(Event|string|array $event): void;
+    public function send(string|array|object $event): void;
 
     /**
      * 发送注释帧（常用于心跳）。

tests/sse/CgiEmitterTest.php

@@ -36,11 +36,13 @@ public function testSendAcceptsStringArrayAndEvent(): void
 
         $emitter->send('hello');
         $emitter->send(['name' => 'fastd']);
+        $emitter->send((object) ['source' => 'object']);
         $emitter->send(new Event('done', 'done', '9'));
 
         $this->assertSame([
             "data: hello\n\n",
             "data: {\"name\":\"fastd\"}\n\n",
+            "data: {\"source\":\"object\"}\n\n",
             "id: 9\nevent: done\ndata: done\n\n",
         ], $this->frames);
     }

tests/sse/SseResponseTest.php

@@ -76,6 +76,21 @@ public function testClosureProducerReceivesEmitter(): void
         $this->assertSame([": connected\n\n", "event: notice\ndata: hello\n\n"], $emitter->frames);
     }
 
+    public function testEmitUsesInjectedEmitterAndRetry(): void
+    {
+        $response = new SseResponse(
+            function (EmitterInterface $emitter): void {
+                $emitter->send('hello');
+            },
+            ['retry' => 1500],
+        );
+
+        $emitter = $this->collectingEmitter();
+        $response->emit($emitter);
+
+        $this->assertSame(["retry: 1500\n\n", "data: hello\n\n"], $emitter->frames);
+    }
+
     public function testOptionsAreExposed(): void
     {
         $response = new SseResponse([], [
@@ -159,7 +174,7 @@ private function collectingEmitter(): EmitterInterface
             /** @var list<string> */
             public array $frames = [];
 
-            public function send(Event|string|array $event): void
+            public function send(string|array|object $event): void
             {
                 $this->frames[] = ($event instanceof Event ? $event : Event::data($event))->format();
             }