需求背景#

1
<?php
2
namespace plugin\webman\gateway;
3

4
use GatewayWorker\Lib\Gateway;
5

6
class Events
7
{
8
    public static function onWorkerStart($worker) {}
9

10
    public static function onConnect($client_id)
11
    {
12
        // 只发送 client_id，让前端通过 HTTP 绑定
13
        Gateway::sendToClient($client_id, json_encode([
14
            'type'      => 'init',
15
            'client_id' => $client_id
16
        ]));
17
    }
18

19
    public static function onWebSocketConnect($client_id, $data) {}
20
    public static function onMessage($client_id, $message) {}
21
    public static function onClose($client_id) {}
22
}

1
<?php
2
namespace app\module\System;
3

4
use app\module\BaseModule;
5
use app\validate\System\WebSocketValidate;
6
use GatewayWorker\Lib\Gateway;
7

8
class WebSocketModule extends BaseModule
9
{
10
    public function __construct()
11
    {
12
        Gateway::$registerAddress = '127.0.0.1:1236';
13
    }
14

15
    public function bind($request): array
16
    {
17
        $param = $this->validate($request, WebSocketValidate::bind());
18
        $userId = $request->userId;
19

20
        Gateway::bindUid($param['client_id'], $userId);
21
        Gateway::sendToClient($param['client_id'], json_encode([
22
            'type' => 'bind_success',
23
            'data' => ['uid' => $userId]
24
        ]));
25

26
        \support\Log::info("[WebSocket] 用户上线: uid={$userId}");
27
        return self::success(['bound' => true]);
28
    }
29

30
    public function pushToUser($request): array
31
    {
32
        $param = $this->validate($request, WebSocketValidate::pushToUser());
33

34
        Gateway::sendToUid($param['uid'], json_encode([
35
            'type' => $param['type'] ?? 'notification',
36
            'data' => $param['data'] ?? []
37
        ]));
38

39
        return self::success(['sent' => true]);
40
    }
41
}

1
export function useWebSocket(options: UseWebSocketOptions = {}) {
2
  const ws = ref<WebSocket | null>(null)
3
  const isConnected = ref(false)
4
  const isBound = ref(false)
5
  const clientId = ref('')
6

7
  function connect() {
8
    ws.value = new WebSocket(wsUrl)
9

10
    ws.value.onopen = () => {
11
      console.log('[WebSocket] 连接成功')
12
      isConnected.value = true
13
    }
14

15
    ws.value.onmessage = (event) => {
16
      const message = JSON.parse(event.data)
17
      handleMessage(message)
18
    }
19
  }
20

21
  function handleMessage(message: WsMessage) {
22
    switch (message.type) {
23
      case 'init':
24
        clientId.value = message.client_id
25
        bindUser()  // 通过 HTTP 接口绑定
26
        break
27
      case 'bind_success':
28
        console.log('[WebSocket] 绑定成功')
29
        isBound.value = true
30
        break
31
    }
32
  }
33

34
  async function bindUser() {
35
    await request.post('/api/admin/WebSocket/bind', {
36
      client_id: clientId.value
37
    })
38
  }
39

40
  onMounted(() => connect())
41
  onUnmounted(() => disconnect())
42

43
  return { isConnected, isBound, send }
44
}

1
// NotificationCenter.vue - 通知组件内监听
2
import { onWsMessage } from '@/hooks/useWebSocket'
3

4
let unsubscribe: (() => void) | null = null
5

6
onMounted(() => {
7
  unsubscribe = onWsMessage('notification', ({ data }) => {
8
    unreadCount.value++
9
    if (data.level === 'urgent') {
10
      ElNotification({ title: data.title, message: data.content, type: data.notification_type })
11
    }
12
  })
13
})
14

15
onUnmounted(() => unsubscribe?.())

1
const HEARTBEAT_INTERVAL = 30000 // 30秒
2
const HEARTBEAT_TIMEOUT = 10000  // 10秒无响应视为断开
3

4
let heartbeatTimer: ReturnType<typeof setInterval> | null = null
5
let heartbeatTimeoutTimer: ReturnType<typeof setTimeout> | null = null
6

7
function startHeartbeat() {
8
  heartbeatTimer = setInterval(() => {
9
    if (ws.value?.readyState === WebSocket.OPEN) {
10
      ws.value.send(JSON.stringify({ type: 'ping' }))
11

12
      // 设置超时检测
13
      heartbeatTimeoutTimer = setTimeout(() => {
14
        console.warn('[WebSocket] 心跳超时，准备重连')
15
        ws.value?.close()
16
        reconnect()
17
      }, HEARTBEAT_TIMEOUT)
18
    }
19
  }, HEARTBEAT_INTERVAL)
20
}
21

22
// 收到 pong 时清除超时计时器
23
function handlePong() {
24
  if (heartbeatTimeoutTimer) {
25
    clearTimeout(heartbeatTimeoutTimer)
26
    heartbeatTimeoutTimer = null
27
  }
28
}

1
public static function onMessage($client_id, $message)
2
{
3
    $data = json_decode($message, true);
4
    if ($data['type'] === 'ping') {
5
        Gateway::sendToClient($client_id, json_encode(['type' => 'pong']));
6
    }
7
}

1
WebSocket 连接 → 获得 client_id → HTTP POST /bind (带 Token) → 服务端验证 Token → 绑定 uid

1
public function pushToUser($request): array
2
{
3
    $param = $this->validate($request, WebSocketValidate::pushToUser());
4

5
    // 只允许推送给自己或下级用户
6
    $currentUserId = $request->userId;
7
    $targetUserId = $param['uid'];
8

9
    if ($currentUserId !== $targetUserId) {
10
        $hasPermission = $this->dep(UsersDep::class)
11
            ->isSubordinate($currentUserId, $targetUserId);
12
        self::throwUnless($hasPermission, '无权向该用户推送消息');
13
    }
14

15
    Gateway::sendToUid($targetUserId, json_encode([
16
        'type' => $param['type'] ?? 'notification',
17
        'data' => $param['data'] ?? [],
18
    ]));
19

20
    return self::success(['sent' => true]);
21
}