| | |
| | | import com.ard.sample.component.websocket.enums.MessageType; |
| | | import com.ard.sample.component.websocket.model.Payload; |
| | | import com.ard.sample.component.websocket.model.WebSocketMessage; |
| | | import com.ard.sample.manage.service.IControlService; |
| | | import lombok.extern.slf4j.Slf4j; |
| | | import org.springframework.beans.factory.annotation.Autowired; |
| | | import org.springframework.stereotype.Component; |
| | |
| | | |
| | | @Autowired |
| | | private LiveStreamHandler _liveStreamService; |
| | | |
| | | @Autowired |
| | | private IControlService _controlService; |
| | | |
| | | // ====================== 心跳相关常量(可配置化建议放到配置文件) ====================== |
| | | /** 心跳间隔(秒):客户端需每隔此时间发送一次PING消息 */ |
| | |
| | | // 3. 处理订阅/取消订阅逻辑 |
| | | handleSubscription(userId, messageType, payload); |
| | | |
| | | // 直播类型订阅:触发开启直播 |
| | | if (MessageType.LIVESTREAM.getType().equals(messageType) && payload.isSubscribe()) { |
| | | String sn = payload.getSn(); |
| | | _liveStreamService.StartLiveStreaming(sn,userId); |
| | | log.info("设备[{}]直播流已开启(用户[{}]订阅触发)", sn, userId); |
| | | } |
| | | |
| | | } catch (Exception e) { |
| | | log.error("用户[{}]消息处理异常", userId, e); |
| | | sendMessage(session, "消息格式错误或处理失败:" + e.getMessage()); |
| | |
| | | subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId); |
| | | // log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type); |
| | | |
| | | // 直播类型订阅:触发开启直播 |
| | | if (MessageType.LIVESTREAM.getType().equals(type)) { |
| | | _liveStreamService.StartLiveStreaming(sn); |
| | | log.info("设备[{}]直播流已开启(用户[{}]订阅触发)", sn, userId); |
| | | } |
| | | // // 直播类型订阅:触发开启直播 |
| | | // if (MessageType.LIVESTREAM.getType().equals(type)) { |
| | | // _liveStreamService.StartLiveStreaming(sn); |
| | | // log.info("设备[{}]直播流已开启(用户[{}]订阅触发)", sn, userId); |
| | | // } |
| | | } else { |
| | | // 取消订阅操作 |
| | | if (subscriptions.containsKey(subscriptionKey)) { |
| | |
| | | * @param type 消息类型(对应MessageType) |
| | | * @param content 消息内容(JSON字符串) |
| | | */ |
| | | public void sendMessageToSubscribers(String sn, String type, String content) { |
| | | public void sendMessageToSubscribers(String sn, String type, String content, String uId) { |
| | | String subscriptionKey = getSubscriptionKey(sn, type); |
| | | Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, Collections.emptySet()); |
| | | |
| | |
| | | |
| | | // 线程安全发送消息 |
| | | try { |
| | | if(MessageType.LIVESTREAM.getType().equals(type)) { |
| | | WebSocketSession liveSession = userSessions.get(uId); |
| | | sendMessage(liveSession, content); |
| | | log.info("单独向用户[{}]推送设备[{}]的[{}]类型消息:{}", uId, sn, type, content); |
| | | return; |
| | | } |
| | | sendMessage(session, content); |
| | | log.debug("向用户[{}]推送设备[{}]的[{}]类型消息:{}", userId, sn, type, content); |
| | | } catch (IOException e) { |