Added method Sender::close(int $fd, bool $reset = null). (#1920)

Author: limingxinleo

src/Listener/OnPipeMessageListener.php

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

src/Sender.php

@@ -21,6 +21,7 @@
 
 /**
  * @method push(int $fd, $data, int $opcode = null, $finish = null)
+ * @method close(int $fd, bool $reset = null)
  */
 class Sender
 {
@@ -60,29 +61,29 @@ public function __construct(ContainerInterface $container)
 
     public function __call($name, $arguments)
     {
-        $fd = $this->getFdFromProxyMethod($name, $arguments);
+        [$fd, $method] = $this->getFdAndMethodFromProxyMethod($name, $arguments);
 
         if ($this->isCoroutineServer) {
             if (isset($this->responses[$fd])) {
                 array_shift($arguments);
-                $this->responses[$fd]->push(...$arguments);
+                $this->responses[$fd]->{$method}(...$arguments);
                 $this->logger->debug("[WebSocket] Worker send to #{$fd}");
             }
             return;
         }
 
-        if (! $this->proxy($fd, $arguments)) {
+        if (! $this->proxy($fd, $method, $arguments)) {
             $this->sendPipeMessage($name, $arguments);
         }
     }
 
-    public function proxy(int $fd, array $arguments): bool
+    public function proxy(int $fd, string $method, array $arguments): bool
     {
         $result = $this->check($fd);
         if ($result) {
             /** @var \Swoole\WebSocket\Server $server */
             $server = $this->getServer();
-            $server->push(...$arguments);
+            $server->{$method}(...$arguments);
             $this->logger->debug("[WebSocket] Worker.{$this->workerId} send to #{$fd}");
         }
 
@@ -114,13 +115,16 @@ public function setResponse(int $fd, ?Response $response): void
         }
     }
 
-    public function getFdFromProxyMethod(string $method, array $arguments): int
+    public function getFdAndMethodFromProxyMethod(string $method, array $arguments): array
     {
-        if (! in_array($method, ['push', 'send', 'sendto'])) {
+        if (! in_array($method, ['push', 'send', 'sendto', 'close'])) {
             throw new InvalidMethodException(sprintf('Method [%s] is not allowed.', $method));
         }
 
-        return (int) $arguments[0];
+        if ($method !== 'close') {
+            $method = 'push';
+        }
+        return [(int) $arguments[0], $method];
     }
 
     protected function getServer(): Server

src/Server.php

@@ -170,6 +170,7 @@ public function onHandShake(SwooleRequest $request, SwooleResponse $response): v
                 if ($server instanceof \Swoole\Coroutine\Http\Server) {
                     $response->upgrade();
                     $this->getSender()->setResponse($fd, $response);
+                    $this->deferOnOpen($request, $class, $response);
 
                     [, , $callbacks] = ServerManager::get($this->serverName);
 
@@ -192,12 +193,7 @@ public function onHandShake(SwooleRequest $request, SwooleResponse $response): v
                         $onMessageCallbackInstance->{$onMessageCallbackMethod}($response, $frame);
                     }
                 } else {
-                    defer(function () use ($request, $class, $server) {
-                        $instance = $this->container->get($class);
-                        if ($instance instanceof OnOpenInterface) {
-                            $instance->onOpen($server, $request);
-                        }
-                    });
+                    $this->deferOnOpen($request, $class, $server);
                 }
             }
         } catch (\Throwable $throwable) {
@@ -257,6 +253,19 @@ public function onClose($server, int $fd, int $reactorId): void
         }
     }
 
+    /**
+     * @param SwooleResponse|WebSocketServer $server
+     */
+    protected function deferOnOpen(SwooleRequest $request, string $class, $server)
+    {
+        defer(function () use ($request, $class, $server) {
+            $instance = $this->container->get($class);
+            if ($instance instanceof OnOpenInterface) {
+                $instance->onOpen($server, $request);
+            }
+        });
+    }
+
     /**
      * Initialize PSR-7 Request.
      */