支持ack

top-think · Jan 19, 2021 · b2620a9 · b2620a9
1 parent 0cfb71c
commit b2620a9
Show file tree

Hide file tree

Showing 4 changed files with 48 additions and 67 deletions.
diff --git a/src/concerns/InteractsWithWebsocket.php b/src/concerns/InteractsWithWebsocket.php
@@ -9,11 +9,9 @@
 use think\Container;
 use think\helper\Str;
 use think\Pipeline;
-use think\swoole\contract\websocket\HandlerInterface;
 use think\swoole\contract\websocket\RoomInterface;
 use think\swoole\Websocket;
 use think\swoole\websocket\Room;
-use think\swoole\websocket\socketio\Handler;
 
 /**
  * Trait InteractsWithWebsocket
@@ -190,7 +188,8 @@ protected function prepareWebsocketListener()
      */
     protected function bindWebsocketHandler()
     {
-        if (($handlerClass = $this->getConfig('websocket.handler')) && $handlerClass instanceof Websocket) {
+        $handlerClass = $this->getConfig('websocket.handler');
+        if ($handlerClass && is_subclass_of($handlerClass, Websocket::class)) {
             $this->app->bind(Websocket::class, $handlerClass);
         }
     }

diff --git a/src/config/swoole.php b/src/config/swoole.php
@@ -1,7 +1,6 @@
 <?php
 
 use think\swoole\websocket\socketio\Handler;
-use think\swoole\websocket\socketio\Parser;
 
 return [
     'server'     => [

diff --git a/src/websocket/socketio/Handler.php b/src/websocket/socketio/Handler.php
@@ -47,6 +47,25 @@ public function onOpen($fd, Request $request)
         if ($this->server->isEstablished($fd)) {
             $this->server->push($fd, $initPayload);
         }
+        if ($this->eio < 4) {
+            $this->onConnect($fd);
+        }
+    }
+
+    protected function onConnect($fd, $data = null)
+    {
+        try {
+            $this->event->trigger('swoole.websocket.Connect', $data);
+            $payload = Packet::MESSAGE . Packet::CONNECT;
+            if ($this->eio >= 4) {
+                $payload .= json_encode(['sid' => base64_encode(uniqid())]);
+            }
+        } catch (Exception $exception) {
+            $payload = sprintf(Packet::MESSAGE . Packet::CONNECT_ERROR . '"%s"', $exception->getMessage());
+        }
+        if ($this->server->isEstablished($fd)) {
+            $this->server->push($fd, $payload);
+        }
     }
 
     /**
@@ -61,24 +80,25 @@ public function onMessage(Frame $frame)
 
         switch ($packet->getEngineType()) {
             case Packet::MESSAGE:
+                $payload = substr($packet->getPayload(), 1);
                 switch ($packet->getSocketType()) {
                     case Packet::CONNECT:
-                        try {
-                            $this->event->trigger('swoole.websocket.Connect');
-                            $payload = Packet::MESSAGE . Packet::CONNECT;
-                            if ($this->eio >= 4) {
-                                $payload .= json_encode(['sid' => base64_encode(uniqid())]);
-                            }
-                        } catch (Exception $exception) {
-                            $payload = sprintf(Packet::MESSAGE . Packet::CONNECT_ERROR . '"%s"', $exception->getMessage());
-                        }
-                        if ($this->server->isEstablished($frame->fd)) {
-                            $this->server->push($frame->fd, $payload);
-                        }
+                        $this->onConnect($frame->fd, $payload);
                         break;
                     case Packet::EVENT:
-                        $payload = substr($packet->getPayload(), 1);
-                        $this->event->trigger('swoole.websocket.Event', $this->decode($payload));
+                    case Packet::ACK:
+                        $start = strpos($payload, '[');
+
+                        if ($start > 0) {
+                            $id      = substr($payload, 0, $start);
+                            $payload = substr($payload, $start);
+                        }
+
+                        $result = $this->event->trigger('swoole.websocket.Event', $this->decode($payload));
+
+                        if (isset($id)) {
+                            $this->server->push($frame->fd, $this->pack(Packet::ACK . $id, end($result)));
+                        }
                         break;
                 }
                 break;
@@ -114,13 +134,19 @@ protected function decode($payload)
         ];
     }
 
-    protected function encode(string $event, $data)
+    protected function pack($type, ...$args)
     {
-        $packet       = Packet::MESSAGE . Packet::EVENT;
-        $shouldEncode = is_array($data) || is_object($data);
-        $data         = $shouldEncode ? json_encode($data) : $data;
-        $format       = $shouldEncode ? '["%s",%s]' : '["%s","%s"]';
+        $packet = Packet::MESSAGE . $type;
 
-        return $packet . sprintf($format, $event, $data);
+        $data = implode(",", array_map(function ($arg) {
+            return json_encode($arg);
+        }, $args));
+
+        return "{$packet}[{$data}]";
+    }
+
+    protected function encode(string $event, $data)
+    {
+        return $this->pack(Packet::EVENT, $event, $data);
     }
 }
diff --git a/src/websocket/socketio/Parser.php b/src/websocket/socketio/Parser.php