Sender support coroutine server. (#1919)

* Sender support coroutine server.

* fixed travis.

* Update OnPipeMessageListener.php

* Update Server.php

Author: limingxinleo

src/Listener/OnPipeMessageListener.php

@@ -64,7 +64,8 @@ public function process(object $event)
             $message = $event->data;
 
             try {
-                $this->sender->proxy($message->name, $message->arguments);
+                $fd = $this->sender->getFdFromProxyMethod($message->name, $message->arguments);
+                $this->sender->proxy($fd, $message->arguments);
             } catch (\Throwable $exception) {
                 $formatter = $this->container->get(FormatterInterface::class);
                 $this->logger->warning($formatter->format($exception));

src/Sender.php

@@ -11,9 +11,12 @@
  */
 namespace Hyperf\WebSocketServer;
 
+use Hyperf\Contract\ConfigInterface;
 use Hyperf\Contract\StdoutLoggerInterface;
+use Hyperf\Server\CoroutineServer;
 use Hyperf\WebSocketServer\Exception\InvalidMethodException;
 use Psr\Container\ContainerInterface;
+use Swoole\Http\Response;
 use Swoole\Server;
 
 /**
@@ -36,23 +39,45 @@ class Sender
      */
     protected $workerId;
 
+    /**
+     * @var Response[]
+     */
+    protected $responses = [];
+
+    /**
+     * @var bool
+     */
+    protected $isCoroutineServer = false;
+
     public function __construct(ContainerInterface $container)
     {
         $this->container = $container;
         $this->logger = $container->get(StdoutLoggerInterface::class);
+        if ($config = $container->get(ConfigInterface::class)) {
+            $this->isCoroutineServer = $config->get('server.type') === CoroutineServer::class;
+        }
     }
 
     public function __call($name, $arguments)
     {
-        if (! $this->proxy($name, $arguments)) {
+        $fd = $this->getFdFromProxyMethod($name, $arguments);
+
+        if ($this->isCoroutineServer) {
+            if (isset($this->responses[$fd])) {
+                array_shift($arguments);
+                $this->responses[$fd]->push(...$arguments);
+                $this->logger->debug("[WebSocket] Worker send to #{$fd}");
+            }
+            return;
+        }
+
+        if (! $this->proxy($fd, $arguments)) {
             $this->sendPipeMessage($name, $arguments);
         }
     }
 
-    public function proxy(string $name, array $arguments): bool
+    public function proxy(int $fd, array $arguments): bool
     {
-        $fd = $this->getFdFromProxyMethod($name, $arguments);
-
         $result = $this->check($fd);
         if ($result) {
             /** @var \Swoole\WebSocket\Server $server */
@@ -80,7 +105,16 @@ public function check($fd): bool
         return false;
     }
 
-    protected function getFdFromProxyMethod(string $method, array $arguments): int
+    public function setResponse(int $fd, ?Response $response): void
+    {
+        if ($response === null) {
+            unset($this->responses[$fd]);
+        } else {
+            $this->responses[$fd] = $response;
+        }
+    }
+
+    public function getFdFromProxyMethod(string $method, array $arguments): int
     {
         if (! in_array($method, ['push', 'send', 'sendto'])) {
             throw new InvalidMethodException(sprintf('Method [%s] is not allowed.', $method));

src/Server.php

@@ -127,6 +127,11 @@ public function getServer()
         return $this->container->get(SwooleServer::class);
     }
 
+    public function getSender(): Sender
+    {
+        return $this->container->get(Sender::class);
+    }
+
     public function onHandShake(SwooleRequest $request, SwooleResponse $response): void
     {
         try {
@@ -164,7 +169,9 @@ public function onHandShake(SwooleRequest $request, SwooleResponse $response): v
                 $server = $this->getServer();
                 if ($server instanceof \Swoole\Coroutine\Http\Server) {
                     $response->upgrade();
-                    [,,$callbacks] = ServerManager::get($this->serverName);
+                    $this->getSender()->setResponse($fd, $response);
+
+                    [, , $callbacks] = ServerManager::get($this->serverName);
 
                     [$onMessageCallbackClass, $onMessageCallbackMethod] = $callbacks[SwooleEvent::ON_MESSAGE];
                     $onMessageCallbackInstance = $this->container->get($onMessageCallbackClass);
@@ -197,6 +204,7 @@ public function onHandShake(SwooleRequest $request, SwooleResponse $response): v
             // Delegate the exception to exception handler.
             $psr7Response = $this->exceptionHandlerDispatcher->dispatch($throwable, $this->exceptionHandlers);
         } finally {
+            isset($fd) && $this->getSender()->setResponse($fd, null);
             // Send the Response to client.
             if (! $psr7Response || ! $psr7Response instanceof Psr7Response) {
                 return;