sse支持

Author: yunwuxin

src/concerns/InteractsWithHttp.php

@@ -2,16 +2,18 @@
 
 namespace think\worker\concerns;
 
+use Stringable;
 use think\App;
 use think\Cookie;
 use think\Event;
 use think\exception\Handle;
 use think\helper\Str;
 use think\Http;
 use think\response\View;
-use think\worker\response\File as FileResponse;
 use think\worker\App as WorkerApp;
 use think\worker\Http as WorkerHttp;
+use think\worker\response\File as FileResponse;
+use think\worker\response\Iterator as IteratorResponse;
 use Throwable;
 use Workerman\Connection\TcpConnection;
 use Workerman\Protocols\Http\Chunk;
@@ -105,6 +107,9 @@ public function onRequest(TcpConnection $connection, WorkerRequest $wkRequest)
 
             $this->sendResponse($connection, $request, $response, $app->cookie);
 
+            //关闭连接
+            $connection->close();
+
             $http->end($response);
         });
     }
@@ -169,6 +174,9 @@ protected function prepareResponse(\think\Response $response)
     protected function sendResponse(TcpConnection $connection, \think\Request $request, \think\Response $response, Cookie $cookie)
     {
         switch (true) {
+            case $response instanceof IteratorResponse:
+                $this->sendIterator($connection, $response, $cookie);
+                break;
             case $response instanceof FileResponse:
                 $this->sendFile($connection, $request, $response, $cookie);
                 break;
@@ -177,6 +185,26 @@ protected function sendResponse(TcpConnection $connection, \think\Request $reque
         }
     }
 
+    protected function sendIterator(TcpConnection $connection, IteratorResponse $response, Cookie $cookie)
+    {
+        $wkResponse = $this->createResponse($response, $cookie);
+        $connection->send($wkResponse);
+
+        foreach ($response as $content) {
+            $chunk = new class($content) implements Stringable {
+                public function __construct(protected $content)
+                {
+                }
+
+                public function __toString(): string
+                {
+                    return $this->content;
+                }
+            };
+            $connection->send($chunk);
+        }
+    }
+
     protected function sendFile(TcpConnection $connection, \think\Request $request, FileResponse $response, Cookie $cookie)
     {
         $ifNoneMatch = $request->header('If-None-Match');
@@ -263,12 +291,12 @@ protected function sendContent(TcpConnection $connection, \think\Response $respo
         $connection->send(new Chunk(''));
     }
 
-    protected function createResponse(\think\Response $response, Cookie $cookie)
+    protected function createResponse(\think\Response $response, Cookie $cookie, $body = '')
     {
         $code   = $response->getCode();
         $header = $response->getHeader();
 
-        $wkResponse = new Response($code, $header);
+        $wkResponse = new Response($code, $header, $body);
 
         foreach ($cookie->getCookie() as $name => $val) {
             [$value, $expire, $option] = $val;

src/response/Iterator.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace think\worker\response;
+
+use IteratorAggregate;
+use think\Response;
+use Traversable;
+
+class Iterator extends Response implements IteratorAggregate
+{
+    protected $iterator;
+
+    public function __construct(Traversable $iterator)
+    {
+        $this->iterator = $iterator;
+    }
+
+    public function getIterator(): Traversable
+    {
+        return $this->iterator;
+    }
+}

tests/feature/HttpTest.php

@@ -12,7 +12,7 @@
 
     while (!$process->getOutput()) {
         $wait++;
-        if ($wait > 10) {
+        if ($wait > 30) {
             throw new Exception('server start failed');
         }
         sleep(1);
@@ -95,6 +95,30 @@
         ->toBe(file_get_contents(STUB_DIR . '/public/asset.txt'));
 });
 
+it('sse', function () {
+    $response = $this->httpClient->get('/sse', [
+        'stream'  => true,
+        'timeout' => 3,
+    ]);
+
+    $body = $response->getBody();
+
+    $buffer = '';
+    while (!$body->eof()) {
+        $text = $body->read(1);
+        if ($text == "\r") {
+            continue;
+        }
+        $buffer .= $text;
+        if ($text == "\n") {
+            if ($buffer != "\n") {
+                expect($buffer)->toStartWith('data: ');
+            }
+            $buffer = '';
+        }
+    }
+});
+
 it('hot update', function () {
     $response = $this->httpClient->get('/hot');
 

tests/stub/config/worker.php

@@ -14,7 +14,7 @@
         'enable'     => env('HTTP_ENABLE', true),
         'host'       => '0.0.0.0',
         'port'       => 8080,
-        'worker_num' => 2,
+        'worker_num' => 1,
         'options'    => [],
     ],
     //队列

tests/stub/route/app.php

@@ -14,6 +14,24 @@
     return 'delete';
 });
 
+Route::get('/sse', function () {
+
+    $generator = function () {
+        foreach (range(0, 9) as $event) {
+            yield 'data: ' . json_encode($event) . "\n\n";
+        }
+
+        yield "data: [DONE]\n\n";
+    };
+
+    $response = new \think\worker\response\Iterator($generator());
+
+    return $response->header([
+        'Content-Type'  => 'text/event-stream',
+        'Cache-Control' => 'no-cache, must-revalidate',
+    ]);
+});
+
 Route::get('test', 'index/test');
 Route::post('json', 'index/json');