easyswoole实现websocket

2021-06-18 php swoole, easyswoole, websocket

easyswoole实现websocket的步骤解析

配置 Websocket服务

1
2
3
4
5
6
7
'MAIN_SERVER' => [
  'SERVER_TYPE' => EASYSWOOLE_WEB_SOCKET_SERVER,
  'SETTING' => [
    // 该参数项为心跳检测，严格参考swoole 配置说明
    'heartbeat_check_interval' => 60,
  ],
],

fd与uuid绑定类App/Support/FdManager.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
<?php
namespace App\Support;
use EasySwoole\Component\Singleton;
use Swoole\Table;
/**
 * Class FdManager
 * @package App\Support
 */
class FdManager
{
    use Singleton;

    /**
     * @var Table
     */
    private $fdUuid;
    /**
     * @var Table
     */
    private $uuidFd;

    /**
     * FdManager constructor.
     * @param int $size
     */
    public function __construct(int $size = 1024 * 256)
    {
        $this->fdUuid = new Table($size);
        $this->fdUuid->column('uuid', Table::TYPE_STRING, 25);
        $this->fdUuid->create();
        $this->uuidFd = new Table($size);
        $this->uuidFd->column('fd', Table::TYPE_INT, 10);
        $this->uuidFd->create();
    }

    /**
     * fd 绑定
     * @param int $fd
     * @param string $uuid
     * @return mixed
     */
    public function bind(int $fd, $uuid)
    {
        // TODO: Implement bind() method.
        $this->fdUuid->set($fd, ['uuid' => $uuid]);
        $this->uuidFd->set($uuid, ['fd' => $fd]);
    }

    /**
     * 删除fd绑定关系
     * @param int $fd
     * @return mixed
     */
    public function delete(int $fd)
    {
        // TODO: Implement delete() method.
        $uuid = $this->fdUuid($fd);
        if ($uuid) {
            $this->uuidFd->del($uuid);
        }
        $this->fdUuid->del($fd);
    }

    /**
     * 通过fd找到绑定的uuid
     * @param int $fd
     * @return mixed
     */
    public function fdUuid(int $fd)
    {
        // TODO: Implement fdUuid() method.
        $ret = $this->fdUuid->get($fd);
        if ($ret) {
            return $ret['uuid'];
        } else {
            return null;
        }
    }

    /**
     * 通过uuid找到fd
     * @param  $uuid
     * @return mixed
     */
    public function uuidFd($uuid)
    {
        // TODO: Implement uuidFd() method.
        $ret = $this->uuidFd->get($uuid);
        if ($ret) {
            return $ret['fd'];
        } else {
            return null;
        }
    }
}

webscoekt 事件类/App/WebSocket/WebSocketEvent.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
<?php
namespace App\WebSocket;
use App\Support\FdManager;
/**
 * Class WebSocketEvent
 * @package App\WebSocket
 */
class WebSocketEvent
{
    /**
     * 握手事件
     * 所有客户端建立连接时触发的方法
     * @param \swoole_http_request $request
     * @param \swoole_http_response $response
     * @return bool
     */
    public function onHandShake(\swoole_http_request $request, \swoole_http_response $response)
    {
        /** 此处自定义握手规则 返回 false 时中止握手 */
        if (!$this->customHandShake($request, $response)) {
            $response->end();
            return false;
        }
        /** 此处是  RFC规范中的WebSocket握手验证过程 必须执行 否则无法正确握手 */
        if ($this->secWebsocketAccept($request, $response)) {
            $response->end();
            return true;
        }
        $response->end();
        return false;
    }

    /**
     * 关闭事件
     * 所有客户端关闭时触发的方法
     * @param \swoole_server $server
     * @param int $fd
     * @param int $reactorId
     */
    public function onClose(\swoole_server $server, int $fd, int $reactorId)
    {
        /** @var array $info */
        $info = $server->getClientInfo($fd);
        /**
         * 判断此fd 是否是一个有效的 websocket 连接
         * 参见 https://wiki.swoole.com/wiki/page/490.html
         */
        if ($info && $info['websocket_status'] === WEBSOCKET_STATUS_FRAME) {
            /**
             * 判断连接是否是 server 主动关闭
             * 参见 https://wiki.swoole.com/wiki/page/p-event/onClose.html
             */
            if ($reactorId < 0) {
            }
            //删除绑定关系
            FdManager::getInstance()->delete($fd);
        }
    }

    /**
     * RFC规范中的WebSocket握手验证过程
     * 以下内容必须强制使用
     *
     * @param \swoole_http_request $request
     * @param \swoole_http_response $response
     * @return bool
     */
    protected function secWebsocketAccept(\swoole_http_request $request, \swoole_http_response $response): bool
    {
        // ws rfc 规范中约定的验证过程
        if (!isset($request->header['sec-websocket-key'])) {
            // 需要 Sec-WebSocket-Key 如果没有拒绝握手
            return false;
        }
        if (0 === preg_match('#^[+/0-9A-Za-z]{21}[AQgw]==$#', $request->header['sec-websocket-key'])
            || 16 !== strlen(base64_decode($request->header['sec-websocket-key']))
        ) {
            //不接受握手
            return false;
        }
        $key = base64_encode(sha1($request->header['sec-websocket-key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true));
        $headers = array(
            'Upgrade' => 'websocket',
            'Connection' => 'Upgrade',
            'Sec-WebSocket-Accept' => $key,
            'Sec-WebSocket-Version' => '13',
            'KeepAlive' => 'off',
        );
        if (isset($request->header['sec-websocket-protocol'])) {
            $headers['Sec-WebSocket-Protocol'] = $request->header['sec-websocket-protocol'];
        }
        // 发送验证后的header
        foreach ($headers as $key => $val) {
            $response->header($key, $val);
        }
        // 接受握手 还需要101状态码以切换状态
        $response->status(101);
        // fd 和 uuid 进行绑定
        FdManager::getInstance()->bind($request->fd, $request->get['uuid']);
        return true;
    }
    
    /**
     * 自定义握手事件
     * @param \swoole_http_request $request
     * @param \swoole_http_response $response
     * @return bool
     */
    private function customHandShake(\swoole_http_request $request, \swoole_http_response $response)
    {
        return true;
    }
}

Webcoket解析类App/WebSocket/WebSocketParser.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
<?php


namespace App\WebSocket;
use EasySwoole\Socket\AbstractInterface\ParserInterface;
use EasySwoole\Socket\Bean\Caller;
use EasySwoole\Socket\Bean\Response;

class WebSocketParser implements ParserInterface
{
    public function decode($raw, $client): ?Caller
    {
        // TODO: Implement decode() method.
        $caller = new Caller;
        if ($raw !== 'PING') {
            $payload = json_decode($raw, true);
            $class = isset($payload['controller']) ? $payload['controller'] : 'index';
            $action = isset($payload['action']) ? $payload['action'] : 'actionNotFound';
            $params = isset($payload['params']) ? (array)$payload['params'] : [];
            $controllerClass = "\\App\\WebSocket\\Controller\\" . ucfirst($class);
            if (!class_exists($controllerClass)) $controllerClass = "\\App\\WebSocket\\Controller\\Index";
            $caller->setClient($caller);
            $caller->setControllerClass($controllerClass);
            $caller->setAction($action);
            $caller->setArgs($params);
        } else {
            // 设置心跳执行的类和方法
            $caller->setControllerClass(\App\WebSocket\Controller\Base::class);
            $caller->setAction('heartbeat');
        }
        return $caller;
    }

    public function encode(Response $response, $client): ?string
    {
        // TODO: Implement encode() method.
        return $response->getMessage();
    }
}

注册 websocket服务 EasySwooleEvent.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public static function mainServerCreate(EventRegister $register)
{
        $conf = new \EasySwoole\Socket\Config();
        //设置Dispatcher为WebSocket 模式
        $conf->setType(\EasySwoole\Socket\Config::WEB_SOCKET);
        try {
            $conf->setParser(new \App\WebSocket\WebSocketParser());//设置解析器对象
            $dispatch = new \EasySwoole\Socket\Dispatcher($conf);//创建Dispatcher对象并注入config对象
        } catch (\Exception $e) {
            Log::error($e->getMessage());
        }
        //给server注册相关事件在WebSocket模式下onMessage事件必须注册 并且交给Dispatcher对象处理
        $register->set(EventRegister::onMessage, function (\swoole_websocket_server $server, \swoole_websocket_frame $frame) use ($dispatch) {
            $dispatch->dispatch($server, $frame->data, $frame);
        });
        $websocketEvent = new \App\WebSocket\WebSocketEvent();
        //自定义握手事件
        $register->set(EventRegister::onHandShake, function (\swoole_http_request $request, \swoole_http_response $response) use ($websocketEvent) {
            $websocketEvent->onHandShake($request, $response);
        });
        //自定义关闭事件
        $register->set(EventRegister::onClose, function (\swoole_server $server, int $fd, int $reactorId) use ($websocketEvent) {
            $websocketEvent->onClose($server, $fd, $reactorId);
        });

}

创建控制器目录，让easyswoole 的websocket服务像http服务那样调用方便

基类控制器App/WebSocket/Controller/Base.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
<?php
namespace App\WebSocket\Controller;
use EasySwoole\Socket\AbstractInterface\Controller;
/**
 * Class Base
 * @package App\WebSocket\Controller
 */
class Base extends Controller
{
    /**
     * 心跳执行的方法
     * 该方法建议 迁移到 基类控制器 Base 中
     * 推荐使用 easyswoole 自带的websocket客户端调试
     * http://www.easyswoole.com/wstool.html
     */
    public function heartbeat()
    {
        $this->response()->setMessage('心跳 heartbeat');
    }

    /**
     * @param string|null $actionName
     */
    protected function actionNotFound(?string $actionName)
    {
        $this->response()->setMessage($actionName . ' not find'); // 推送消息
    }
}

测试控制器App/WebSocket/Controller/Test.php

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<?php
namespace App\WebSocket\Controller;
class Test extends Base
{
    public function index()
    {
        $fd = $this->caller()->getClient()->getFd();// 请求用户的fd
        $data = $this->caller()->getArgs(); // 获取请求参数
      //
        \EasySwoole\EasySwoole\ServerManager::getInstance()->getSwooleServer()->worker_id
        
        //发送响应消息
        $this->response()->setMessage("响应消息");
       
        $server = ServerManager::getInstance()->getSwooleServer();
        //推送消息
        $server->push($fd, “要推送的消息”);
    }
}