diff --git a/src/Response/SseResponse.php b/src/Response/SseResponse.php index e4a6882..d080f57 100644 --- a/src/Response/SseResponse.php +++ b/src/Response/SseResponse.php @@ -47,7 +47,7 @@ final class SseResponse extends Text implements StreamingResponseInterface private ?\Closure $onError; /** - * @param iterable|\Closure(EmitterInterface): void $producer + * @param iterable|\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); diff --git a/src/Sse/AbstractEmitter.php b/src/Sse/AbstractEmitter.php new file mode 100644 index 0000000..c510b4d --- /dev/null +++ b/src/Sse/AbstractEmitter.php @@ -0,0 +1,59 @@ +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); + } +} diff --git a/src/Sse/CgiEmitter.php b/src/Sse/CgiEmitter.php index 527acf9..dbdc303 100644 --- a/src/Sse/CgiEmitter.php +++ b/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); - } } diff --git a/src/Sse/EmitterInterface.php b/src/Sse/EmitterInterface.php index 8104ae8..8b7a583 100644 --- a/src/Sse/EmitterInterface.php +++ b/src/Sse/EmitterInterface.php @@ -15,9 +15,9 @@ interface EmitterInterface /** * 发送一条事件。 * - * @param Event|string|array $event string 作为 data;数组/对象 JSON 编码 + * @param Event|string|array|object $event string 作为 data;数组/对象 JSON 编码 */ - public function send(Event|string|array $event): void; + public function send(string|array|object $event): void; /** * 发送注释帧(常用于心跳)。 diff --git a/tests/sse/CgiEmitterTest.php b/tests/sse/CgiEmitterTest.php index 2c5cb4c..195ecfc 100644 --- a/tests/sse/CgiEmitterTest.php +++ b/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); } diff --git a/tests/sse/SseResponseTest.php b/tests/sse/SseResponseTest.php index 2426d0f..9654fb9 100644 --- a/tests/sse/SseResponseTest.php +++ b/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 */ 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(); }