Merge branch 'master' into 2.1-merge

# Conflicts:
#	README.md
#	src/amqp/src/Message/ConsumerMessageInterface.php
#	src/utils/src/Coroutine/Concurrent.php
#	src/websocket-server/src/Server.php

Author: limingxinleo

src/Server.php

@@ -165,37 +165,41 @@ public function onHandShake(SwooleRequest $request, SwooleResponse $response): v
 
             $class = $psr7Response->getAttribute('class');
 
-            if (! empty($class)) {
-                FdCollector::set($fd, $class);
-                $server = $this->getServer();
-                if ($server instanceof \Swoole\Coroutine\Http\Server) {
-                    $response->upgrade();
-                    $this->getSender()->setResponse($fd, $response);
-                    $this->deferOnOpen($request, $class, $response);
-
-                    [, , $callbacks] = ServerManager::get($this->serverName);
-
-                    [$onMessageCallbackClass, $onMessageCallbackMethod] = $callbacks[Event::ON_MESSAGE];
-                    $onMessageCallbackInstance = $this->container->get($onMessageCallbackClass);
-
-                    [$onCloseCallbackClass, $onCloseCallbackMethod] = $callbacks[Event::ON_CLOSE];
-                    $onCloseCallbackInstance = $this->container->get($onCloseCallbackClass);
-
-                    while (true) {
-                        $frame = $response->recv();
-                        if ($frame === false || $frame instanceof CloseFrame || $frame === '') {
-                            $onCloseCallbackInstance->{$onCloseCallbackMethod}($response, $fd, 0);
-                            break;
-                        }
-                        $onMessageCallbackInstance->{$onMessageCallbackMethod}($response, $frame);
+            if (empty($class)) {
+                throw new WebSocketHandeShakeException('WebSocket hande shake failed, because the class does not exists.');
+            }
+
+            FdCollector::set($fd, $class);
+            $server = $this->getServer();
+            if ($server instanceof \Swoole\Coroutine\Http\Server) {
+                $response->upgrade();
+                $this->getSender()->setResponse($fd, $response);
+                $this->deferOnOpen($request, $class, $response);
+
+                [, , $callbacks] = ServerManager::get($this->serverName);
+
+                [$onMessageCallbackClass, $onMessageCallbackMethod] = $callbacks[Event::ON_MESSAGE];
+                $onMessageCallbackInstance = $this->container->get($onMessageCallbackClass);
+
+                [$onCloseCallbackClass, $onCloseCallbackMethod] = $callbacks[Event::ON_CLOSE];
+                $onCloseCallbackInstance = $this->container->get($onCloseCallbackClass);
+
+                while (true) {
+                    $frame = $response->recv();
+                    if ($frame === false || $frame instanceof CloseFrame || $frame === '') {
+                        $onCloseCallbackInstance->{$onCloseCallbackMethod}($response, $fd, 0);
+                        break;
                     }
-                } else {
-                    $this->deferOnOpen($request, $class, $server);
+                    $onMessageCallbackInstance->{$onMessageCallbackMethod}($response, $frame);
                 }
+            } else {
+                $this->deferOnOpen($request, $class, $server);
             }
         } catch (Throwable $throwable) {
             // Delegate the exception to exception handler.
             $psr7Response = $this->exceptionHandlerDispatcher->dispatch($throwable, $this->exceptionHandlers);
+            FdCollector::del($request->fd);
+            WsContext::release($request->fd);
         } finally {
             isset($fd) && $this->getSender()->setResponse($fd, null);
             // Send the Response to client.