liusuyi
2025-03-10 99b201ea344b2b21efebb6bf8b72397714288783
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
package com.ard.tianyu.websocket.handler;
 
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.ard.tianyu.service.control.UavControlService;
import com.ard.tianyu.websocket.domain.BaseMessage;
import com.ard.tianyu.websocket.domain.UavStreamMessage;
import com.ard.tianyu.websocket.domain.UavStateMessage;
import com.ard.tianyu.websocket.utils.WebSocketUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import javax.annotation.Resource;
 
 
@Component
@Slf4j
public class TyMessageHandler extends TextWebSocketHandler {
 
    @Resource
    private UavStreamManager uavStreamManager;
    @Resource
    private UavControlService uavControlService;
    @Resource
    private UavStateManager uavStateManager;
 
 
    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) {
        String userIdStr = (String) session.getAttributes().get("userId");
        Long userId = Long.parseLong(userIdStr);
        String payload = message.getPayload();
        log.info("【cmd】Received message: {}", payload);
        log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size());
 
        BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class);
        switch (baseMessage.getType()) {
            case "stream":
                handleStreamMessage(payload, userId);
                break;
            case "state":
                handleStateMessage(payload, userId);
                break;
        }
    }
 
    /**
     * 处理流消息
     * <p>
     * 该方法解析给定的负载数据,以获取无人机流消息,并根据消息类型决定是否开始或停止流
     *
     * @param payload 消息的负载数据,通常为JSON字符串
     * @param userId  接收消息的用户ID
     */
    private void handleStreamMessage(String payload, Long userId) {
        // 解析JSON负载数据为BaseMessage对象,包含UavStreamMessage类型的有效载荷
        BaseMessage<UavStreamMessage> streamMessage = JSON.parseObject(payload,
                new TypeReference<BaseMessage<UavStreamMessage>>() {
                });
        UavStreamMessage uavStreamMessage = streamMessage.getPayload();
 
        // 检查解析后的UavStreamMessage对象是否非空
        if (uavStreamMessage != null) {
            String boxSn = uavStreamMessage.getBoxSn();
 
            // 根据pullType决定操作类型:1表示开始流,其他值表示停止流
            if (uavStreamMessage.getPullType().equals(1)) {
                uavStreamManager.startPullStream(boxSn, userId);
            } else {
                uavStreamManager.stopPullStream(boxSn, userId);
            }
        }
    }
 
    /**
     * 处理状态消息
     * 根据接收到的消息内容,决定无人机状态信息的订阅或退订
     *
     * @param payload 消息负载,包含订阅信息
     * @param userId  用户ID,用于标识用户
     */
    private void handleStateMessage(String payload, Long userId) {
        // 解析接收到的JSON消息,将其转换为BaseMessage对象
        BaseMessage<UavStateMessage> stateMessage = JSON.parseObject(payload,
                new TypeReference<BaseMessage<UavStateMessage>>() {
                });
        // 获取消息的有效载荷部分
        UavStateMessage uavStateMessage = stateMessage.getPayload();
 
        // 检查有效载荷是否存在
        if (uavStateMessage != null) {
            // 获取无人机盒子的序列号
            String boxSn = uavStateMessage.getBoxSn();
            // 根据消息中的订阅标志,决定是订阅还是退订无人机状态
            Boolean subscribe = uavStateMessage.getSubscribe();
            if (subscribe) {
                // 执行订阅操作
                uavStateManager.subscribeState(boxSn, userId);
            } else {
                // 执行退订操作
                uavStateManager.unsubscribeState(boxSn, userId);
            }
        }
    }
 
 
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        log.info("【cmd】Connected: {}", session.getId());
        String userIdStr = (String) session.getAttributes().get("userId");
        Long userId = Long.parseLong(userIdStr);
        WebSocketUtil.ONLINE_USER_SESSIONS.put(userId, session);
        log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size());
    }
 
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        log.info("【cmd】Disconnected: {}", session.getId());
        String userIdStr = (String) session.getAttributes().get("userId");
        Long userId = Long.parseLong(userIdStr);
        WebSocketUtil.ONLINE_USER_SESSIONS.remove(userId);
        log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size());
        // 停止用户的所有流
        uavStreamManager.stopAllPullStreamByUser(userId);
        if (userId != null) {
            // 释放该用户所有 UAV 控制权
            uavControlService.releaseAllControls(userId);
        }
 
        uavStateManager.removeSubscribeState(userId);
    }
}