diff --git a/src/Response/SseResponse.php b/src/Response/SseResponse.php index d080f57..79ea478 100644 --- a/src/Response/SseResponse.php +++ b/src/Response/SseResponse.php @@ -96,7 +96,21 @@ public function __construct(iterable|\Closure $producer, array $options = []) return; } - $emitter->send($event); + if ($event instanceof Event || is_string($event) || is_array($event)) { + $emitter->send($event); + continue; + } + + $sendObject = [$emitter, 'sendObject']; + + if (is_callable($sendObject)) { + $sendObject($event); + continue; + } + + throw new InvalidArgumentException( + 'The current SSE emitter does not support object events; implement ObjectEmitterInterface.' + ); } }; } diff --git a/src/Sse/AbstractEmitter.php b/src/Sse/AbstractEmitter.php index c510b4d..469825c 100644 --- a/src/Sse/AbstractEmitter.php +++ b/src/Sse/AbstractEmitter.php @@ -8,12 +8,22 @@ * SSE 发送器公共实现。 * * 统一帧归一化、注释帧、重连间隔和关闭状态;具体运行时只需实现 write()。 + * 对象事件通过 ObjectEmitterInterface::sendObject() 提供,保持 send() 的旧签名。 */ -abstract class AbstractEmitter implements EmitterInterface +abstract class AbstractEmitter implements ObjectEmitterInterface { protected bool $closed = false; - public function send(string|array|object $event): void + public function send(Event|string|array $event): void + { + if ($this->isClosed()) { + return; + } + + $this->write($this->normalize($event)->format()); + } + + public function sendObject(object $event): void { if ($this->isClosed()) { return; diff --git a/src/Sse/EmitterInterface.php b/src/Sse/EmitterInterface.php index 8b7a583..7311adc 100644 --- a/src/Sse/EmitterInterface.php +++ b/src/Sse/EmitterInterface.php @@ -15,9 +15,9 @@ interface EmitterInterface /** * 发送一条事件。 * - * @param Event|string|array|object $event string 作为 data;数组/对象 JSON 编码 + * @param Event|string|array $event string 作为 data;数组 JSON 编码 */ - public function send(string|array|object $event): void; + public function send(Event|string|array $event): void; /** * 发送注释帧(常用于心跳)。 diff --git a/src/Sse/ObjectEmitterInterface.php b/src/Sse/ObjectEmitterInterface.php new file mode 100644 index 0000000..bc8305c --- /dev/null +++ b/src/Sse/ObjectEmitterInterface.php @@ -0,0 +1,19 @@ +send('hello'); $emitter->send(['name' => 'fastd']); - $emitter->send((object) ['source' => 'object']); + $emitter->sendObject((object) ['source' => 'object']); $emitter->send(new Event('done', 'done', '9')); $this->assertSame([ diff --git a/tests/sse/SseResponseTest.php b/tests/sse/SseResponseTest.php index 9654fb9..43e1140 100644 --- a/tests/sse/SseResponseTest.php +++ b/tests/sse/SseResponseTest.php @@ -8,6 +8,7 @@ use FastD\Http\Response\StreamingResponseInterface; use FastD\Http\Sse\EmitterInterface; use FastD\Http\Sse\Event; +use FastD\Http\Sse\ObjectEmitterInterface; use InvalidArgumentException; use PHPUnit\Framework\TestCase; use Psr\Http\Message\ResponseInterface; @@ -50,6 +51,7 @@ public function testIterableProducerIsWrapped(): void $response = new SseResponse([ 'a', ['b' => 2], + (object) ['d' => 4], Event::data('c', 'custom', '3'), ]); @@ -59,6 +61,7 @@ public function testIterableProducerIsWrapped(): void $this->assertSame([ "data: a\n\n", "data: {\"b\":2}\n\n", + "data: {\"d\":4}\n\n", "id: 3\nevent: custom\ndata: c\n\n", ], $emitter->frames); } @@ -170,11 +173,16 @@ public function testCloneKeepsProducerAndOptions(): void */ private function collectingEmitter(): EmitterInterface { - return new class implements EmitterInterface { + return new class implements ObjectEmitterInterface { /** @var list */ public array $frames = []; - public function send(string|array|object $event): void + public function send(Event|string|array $event): void + { + $this->frames[] = ($event instanceof Event ? $event : Event::data($event))->format(); + } + + public function sendObject(object $event): void { $this->frames[] = ($event instanceof Event ? $event : Event::data($event))->format(); }