From 8c5435101924d68588448a4c981417df4b4dab1d Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Fri, 21 Feb 2025 11:08:20 +0800
Subject: [PATCH] 整合websocket 统一使用按类型区分
---
ard-work/src/main/java/com/ard/tianyu/websocket/domain/BaseMessage.java | 2
ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlAuthorityDTO.java | 2
/dev/null | 19 --
ard-work/src/main/java/com/ard/tianyu/controller/UavPermissionsController.java | 2
ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java | 11 +
ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java | 11 -
ard-work/src/main/java/com/ard/tianyu/service/impl/UavStreamManager.java | 129 ++++++++++++++++++
ard-work/src/main/java/com/ard/tianyu/service/impl/UavControlManager.java | 156 +++++++++++++++------
ard-work/src/main/java/com/ard/tianyu/websocket/domain/PullStreamDTO.java | 2
ard-work/src/main/java/com/ard/tianyu/aspect/ControlAuthorityAspect.java | 2
ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java | 35 +++++
11 files changed, 288 insertions(+), 83 deletions(-)
diff --git a/ard-work/src/main/java/com/ard/tianyu/aspect/ControlAuthorityAspect.java b/ard-work/src/main/java/com/ard/tianyu/aspect/ControlAuthorityAspect.java
index 7ec6163..22cd4cd 100644
--- a/ard-work/src/main/java/com/ard/tianyu/aspect/ControlAuthorityAspect.java
+++ b/ard-work/src/main/java/com/ard/tianyu/aspect/ControlAuthorityAspect.java
@@ -57,7 +57,7 @@
String boxSn = getBoxSn(arg);
if (boxSn != null) {
//判断无人机控制者是否为当前登录用户
- Long controller = UavControlManager.droneControlMap.get(boxSn);
+ Long controller = UavControlManager.uavControlMap.get(boxSn);
if (controller != null && controller.equals(SecurityUtils.getUserId())) {
return true;
}
diff --git a/ard-work/src/main/java/com/ard/tianyu/controller/UavPermissionsController.java b/ard-work/src/main/java/com/ard/tianyu/controller/UavPermissionsController.java
index a307240..b258a3c 100644
--- a/ard-work/src/main/java/com/ard/tianyu/controller/UavPermissionsController.java
+++ b/ard-work/src/main/java/com/ard/tianyu/controller/UavPermissionsController.java
@@ -52,6 +52,6 @@
@ApiOperationSupport(order = 4)
@GetMapping("/controlRequestMap")
public AjaxResult controlRequestMap(String boxSn) {
- return AjaxResult.success(UavControlManager.droneControlMap.get(boxSn));
+ return AjaxResult.success(UavControlManager.uavControlMap.get(boxSn));
}
}
diff --git a/ard-work/src/main/java/com/ard/tianyu/service/impl/UavControlManager.java b/ard-work/src/main/java/com/ard/tianyu/service/impl/UavControlManager.java
index 908a009..9d87722 100644
--- a/ard-work/src/main/java/com/ard/tianyu/service/impl/UavControlManager.java
+++ b/ard-work/src/main/java/com/ard/tianyu/service/impl/UavControlManager.java
@@ -3,94 +3,156 @@
import com.alibaba.fastjson2.JSON;
import com.ard.common.exception.NoPermissionException;
import com.ard.common.utils.SecurityUtils;
-import com.ard.tianyu.websocket.domain.ControlMessage;
-import com.ard.tianyu.websocket.domain.WsMessage;
+import com.ard.tianyu.websocket.domain.ControlAuthorityDTO;
+import com.ard.tianyu.websocket.domain.BaseMessage;
import com.ard.tianyu.websocket.utils.WebSocketUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 无人机控制管理器
+ *
+ * @author liusuyi
+ * @date 2024年02月21日8:37
+ **/
@Service
@Slf4j
public class UavControlManager {
- // 存储每个无人机的当前控制者
- public static final ConcurrentHashMap<String, Long> droneControlMap = new ConcurrentHashMap<>();
+ private static final String CONTROL_REQUEST_ACTION = "controlRequest";
+ private static final String REQUEST_ACTION = "request";
+ private static final String APPROVE_ACTION = "approve";
+ private static final String REJECT_ACTION = "reject";
+ // 存储每个无人机的当前控制者
+ public static final ConcurrentHashMap<String, Long> uavControlMap = new ConcurrentHashMap<>();
// 存储无人机的控制请求（用于判断哪个控制者正在等待）
public static final ConcurrentHashMap<String, Long> controlRequestMap = new ConcurrentHashMap<>();
- // 发送通知（通过 WebSocket推送）
- public void notifyMessage(String boxSn, String action, Long source, Long target) {
- ControlMessage controlMessage = ControlMessage.builder()
- .boxSn(boxSn)
- .userId(source)
- .action(action)
- .build();
- WsMessage<ControlMessage> wsMessage = new WsMessage<>("controlRequest", controlMessage);
- WebSocketUtil.sendMessage(target.toString(), JSON.toJSONString(wsMessage));
- }
- // 申请控制
+ /**
+ * 申请控制
+ *
+ * @param boxSn 盒子编号
+ */
public boolean requestControl(String boxSn) {
Long requester = SecurityUtils.getUserId();
+ Long controller = uavControlMap.get(boxSn);
// 如果无人机没有控制者，直接控制
- if (droneControlMap.get(boxSn) == null) {
- droneControlMap.put(boxSn, requester);
+ if (controller == null) {
+ uavControlMap.put(boxSn, requester);
+ log.info("控制权已授予给用户: {}", requester);
return true; // 控制成功
}
// 如果无人机已经有控制者,则发送申请给当前控制者
- Long controller = droneControlMap.get(boxSn);
if (!controller.equals(requester)) {
// 则将控制请求加入请求队列
controlRequestMap.put(boxSn, requester);
// 通知当前控制者，有新的控制请求
- notifyMessage(boxSn, "request", requester, controller);
+ notifyMessage(boxSn, REQUEST_ACTION, requester, controller);
+ log.info("控制请求已发送给当前控制者: {}", controller);
} else {
throw new NoPermissionException("当前已经获取控制权，无需重复申请");
}
return false; // 等待当前控制者的同意
}
- // 处理控制请求的同意或拒绝
+ /**
+ * 处理控制请求
+ *
+ * @param boxSn 盒子编号
+ * @param approve 是否同意
+ */
public boolean handleControlResponse(String boxSn, boolean approve) {
Long controller = SecurityUtils.getUserId();
Long requester = controlRequestMap.get(boxSn);
- // 确保请求者是当前的控制者
- if (requester != null && !requester.equals(controller)) {
- if (approve) {
- // 同意控制，将控制权赋予请求者
- droneControlMap.put(boxSn, requester);
- notifyMessage(boxSn, "approve", controller, requester);
- } else {
- notifyMessage(boxSn, "reject", controller, requester);
- }
- // 无论同意还是拒绝，都移除控制请求
- controlRequestMap.remove(boxSn);
- return true;
- }
- else {
+
+ if (requester == null || requester.equals(controller)) {
throw new NoPermissionException("无申请不需要处理控制权");
}
+
+ if (approve) {
+ uavControlMap.put(boxSn, requester);
+ notifyMessage(boxSn, APPROVE_ACTION, controller, requester);
+ log.info("控制权已转移给用户: {}", requester);
+ } else {
+ notifyMessage(boxSn, REJECT_ACTION, controller, requester);
+ log.info("控制请求被拒绝，请求者: {}", requester);
+ }
+
+ controlRequestMap.remove(boxSn);
+ return true;
}
- // 释放控制
+ /**
+ * 释放控制
+ *
+ * @param boxSn 盒子编号
+ */
public boolean releaseControl(String boxSn) {
- Long controller = droneControlMap.get(boxSn);
- if (controller != null) {
- if (controller.equals(SecurityUtils.getUserId())) {
- droneControlMap.remove(boxSn);
- return true;
- } else {
- // 当前控制者不匹配，可以记录日志或者抛出异常
- log.warn("无权限释放控制, 实际控制者是 {}", controller);
- throw new NoPermissionException("无权限释放控制, 实际控制者是 " + controller);
- }
- } else {
- // 说明无人机没有控制者
+ Long controller = uavControlMap.get(boxSn);
+
+ if (controller == null) {
log.warn("无人机当前 {} 没有控制者,无需释放", boxSn);
throw new NoPermissionException("无人机" + boxSn + "当前没有控制者,无需释放");
}
+
+ if (!controller.equals(SecurityUtils.getUserId())) {
+ log.warn("无权限释放控制, 实际控制者是 {}", controller);
+ throw new NoPermissionException("无权限释放控制, 实际控制者是 " + controller);
+ }
+
+ uavControlMap.remove(boxSn);
+ log.info("控制权已释放，无人机: {}", boxSn);
+ return true;
+ }
+
+
+ /**
+ * 释放指定用户控制的无人机
+ *
+ * @param userId 用户id
+ */
+ public void releaseAllControlByUser(String userId) {
+ String boxSn = uavControlMap.entrySet().stream()
+ .filter(entry -> entry.getValue().toString().equals(userId))
+ .map(ConcurrentHashMap.Entry::getKey)
+ .findFirst()
+ .orElse(null);
+ if (boxSn != null) {
+ // 用户控制了某个无人机，移除缓存
+ uavControlMap.remove(boxSn);
+ log.info("移除{}控制的无人机: {}", userId, boxSn);
+ }
+ }
+
+
+ /**
+ * 发送消息给指定用户
+ *
+ * @param boxSn 盒子编号
+ * @param action 指令()
+ * @param source 发送方用户ID
+ * @param target 接收方用户ID
+ */
+ private void notifyMessage(String boxSn, String action, Long source, Long target) {
+ ControlAuthorityDTO controlAuthorityDTO = buildControlAuthorityDTO(boxSn, action, source);
+ BaseMessage<ControlAuthorityDTO> baseMessage = buildBaseMessage(CONTROL_REQUEST_ACTION, controlAuthorityDTO);
+ WebSocketUtil.sendMessage(target.toString(), JSON.toJSONString(baseMessage));
+ }
+
+ private ControlAuthorityDTO buildControlAuthorityDTO(String boxSn, String action, Long userId) {
+ return ControlAuthorityDTO.builder()
+ .boxSn(boxSn)
+ .userId(userId)
+ .action(action)
+ .build();
+ }
+
+ private <T> BaseMessage<T> buildBaseMessage(String type, T data) {
+ return new BaseMessage<>(type, data);
}
}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/service/impl/UavStreamManager.java b/ard-work/src/main/java/com/ard/tianyu/service/impl/UavStreamManager.java
new file mode 100644
index 0000000..eb3adc9
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/tianyu/service/impl/UavStreamManager.java
@@ -0,0 +1,129 @@
+package com.ard.tianyu.service.impl;
+
+import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver;
+import com.ard.tianyu.service.ITyService;
+import com.ard.tianyu.websocket.utils.WebSocketUtil;
+import com.ard.vtdu.domain.ArdVtdu;
+import com.ard.vtdu.service.IMediaService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+@Service
+@Slf4j
+public class UavStreamManager {
+
+ private static final String STREAM_LOG_PREFIX = "【stream】";
+ private static final String STREAM_NOT_FOUND_MSG = STREAM_LOG_PREFIX + "Box stream not found: ";
+ private static final String STREAM_CLOSED_MSG = STREAM_LOG_PREFIX + "Stream closed due to no active users: ";
+ private static final String STREAM_IN_USE_MSG = STREAM_LOG_PREFIX + "Stream is still in use, remaining users: ";
+
+ @Resource
+ private ITyService tyService;
+ @Resource
+ private IMediaService mediaService;
+
+ // 盒子拉流地址缓存 key:boxSn value: webrtcUrl
+ public static final ConcurrentMap<String, String> BOX_STREAM_PULL_MAP = new ConcurrentHashMap<>();
+ // 盒子拉流用户集合缓存 key:boxSn value:Set<UserId>
+ public static final ConcurrentMap<String, Set<String>> STREAM_USER_MAP = new ConcurrentHashMap<>();
+
+ /**
+ * 开始拉流
+ *
+ * @param boxSn 盒子序列号
+ * @param userId 用户ID
+ */
+ public void startPullStream(String boxSn, String userId) {
+ STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
+
+ String webrtcUrl = BOX_STREAM_PULL_MAP.get(boxSn);
+ if (webrtcUrl != null) {
+ log.info("{}Already pulled stream: {}", STREAM_LOG_PREFIX, webrtcUrl);
+ sendMessageToUser(userId, webrtcUrl);
+ return;
+ }
+
+ try {
+ BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
+ if (boxLiveUrl == null) {
+ log.warn("{}{}{}", STREAM_LOG_PREFIX, STREAM_NOT_FOUND_MSG, boxSn);
+ sendMessageToUser(userId, STREAM_NOT_FOUND_MSG + boxSn);
+ return;
+ }
+
+ ArdVtdu vtdu = ArdVtdu.builder()
+ .name(boxSn)
+ .mode(0)
+ .isCode(0)
+ .rtspSource(boxLiveUrl.getRtsp())
+ .build();
+ ArdVtdu ardVtdu = mediaService.addPath(vtdu);
+ webrtcUrl = ardVtdu.getWebrtcUrl();
+ BOX_STREAM_PULL_MAP.put(boxSn, webrtcUrl);
+
+ log.info("{}First pulled stream: {}", STREAM_LOG_PREFIX, webrtcUrl);
+ sendMessageToUser(userId, webrtcUrl);
+ } catch (Exception e) {
+ log.error("{}Failed to start pull stream for box: {}", STREAM_LOG_PREFIX, boxSn, e);
+ sendMessageToUser(userId, "Failed to start pull stream due to an internal error.");
+ }
+ }
+
+ /**
+ * 停止拉流
+ *
+ * @param boxSn 盒子序列号
+ * @param userId 用户ID
+ */
+ public void stopPullStream(String boxSn, String userId) {
+ synchronized (this) {
+ Set<String> users = STREAM_USER_MAP.get(boxSn);
+ if (users == null) {
+ log.warn("{}No active users for box: {}", STREAM_LOG_PREFIX, boxSn);
+ sendMessageToUser(userId, "No active users for box: " + boxSn);
+ return;
+ }
+
+ users.remove(userId);
+ if (users.isEmpty()) {
+ mediaService.removePath(boxSn);
+ BOX_STREAM_PULL_MAP.remove(boxSn);
+ STREAM_USER_MAP.remove(boxSn);
+ log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_CLOSED_MSG, boxSn);
+ sendMessageToUser(userId, STREAM_CLOSED_MSG + boxSn);
+ } else {
+ log.info("{}{}{}", STREAM_LOG_PREFIX, STREAM_IN_USE_MSG, users);
+ sendMessageToUser(userId, STREAM_IN_USE_MSG + users);
+ }
+ }
+ }
+ /**
+ * 停止拉流
+ *
+ * @param userId 用户ID
+ */
+ public void stopAllPullStreamByUser(String userId) {
+ synchronized (this) {
+ STREAM_USER_MAP.forEach((boxSn, users) -> stopPullStream(boxSn, userId));
+ }
+ }
+
+ /**
+ * 发送消息给指定用户
+ *
+ * @param userId 用户ID
+ * @param message 消息内容
+ */
+ private void sendMessageToUser(String userId, String message) {
+ try {
+ WebSocketUtil.sendMessage(userId, message);
+ } catch (Exception e) {
+ log.error("{}Failed to send message to user: {}", STREAM_LOG_PREFIX, userId, e);
+ }
+ }
+}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java b/ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java
index 757cdc5..bd751f0 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java
@@ -1,12 +1,12 @@
package com.ard.tianyu.websocket.config;
import com.ard.tianyu.websocket.handler.TyMessageHandler;
-import com.ard.tianyu.websocket.handler.TyStreamHandler;
import com.ard.tianyu.websocket.interceptor.TyHandshakeInterceptor;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
+
import javax.annotation.Resource;
@Configuration
@@ -15,20 +15,15 @@
@Resource
private TyMessageHandler tyMessageHandler;
- @Resource
- private TyStreamHandler tyStreamHandler;
+
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
//添加myWebSocketHandler消息处理对象以及websocket连接地址
- registry.addHandler(tyMessageHandler, "/ws/control/**")
+ registry.addHandler(tyMessageHandler, "/ws/**")
//设置允许跨域访问
.setAllowedOrigins("*")
//添加拦截器可实现用户链接前进行权限校验等操作
.addInterceptors(new TyHandshakeInterceptor());
- // 注册流媒体 WebSocket 处理器
- registry.addHandler(tyStreamHandler, "/ws/stream/**")
- .setAllowedOrigins("*")
- .addInterceptors(new TyHandshakeInterceptor()); // 可选的拦截器
}
}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/domain/WsMessage.java b/ard-work/src/main/java/com/ard/tianyu/websocket/domain/BaseMessage.java
similarity index 89%
rename from ard-work/src/main/java/com/ard/tianyu/websocket/domain/WsMessage.java
rename to ard-work/src/main/java/com/ard/tianyu/websocket/domain/BaseMessage.java
index 5314b6d..af4027e 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/domain/WsMessage.java
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/domain/BaseMessage.java
@@ -7,7 +7,7 @@
@Data
@AllArgsConstructor
@NoArgsConstructor
-public class WsMessage<T> {
+public class BaseMessage<T> {
private String type; // 消息类型
private T payload; // 泛型消息内容
}
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlMessage.java b/ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlAuthorityDTO.java
similarity index 92%
rename from ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlMessage.java
rename to ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlAuthorityDTO.java
index c3ad0a7..f9da173 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlMessage.java
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/domain/ControlAuthorityDTO.java
@@ -9,7 +9,7 @@
@Builder
@AllArgsConstructor
@NoArgsConstructor
-public class ControlMessage {
+public class ControlAuthorityDTO {
private String boxSn; // 盒子编号
private Long userId; // 消息发送者
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java b/ard-work/src/main/java/com/ard/tianyu/websocket/domain/PullStreamDTO.java
similarity index 90%
rename from ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java
rename to ard-work/src/main/java/com/ard/tianyu/websocket/domain/PullStreamDTO.java
index 3e1067d..2f6e074 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/domain/PullStreamDTO.java
@@ -1,4 +1,4 @@
-package com.ard.tianyu.websocket.model;
+package com.ard.tianyu.websocket.domain;
import lombok.AllArgsConstructor;
import lombok.Builder;
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java b/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java
index 6c6f007..d7be5e8 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java
@@ -1,5 +1,11 @@
package com.ard.tianyu.websocket.handler;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.TypeReference;
+import com.ard.tianyu.service.impl.UavControlManager;
+import com.ard.tianyu.service.impl.UavStreamManager;
+import com.ard.tianyu.websocket.domain.BaseMessage;
+import com.ard.tianyu.websocket.domain.PullStreamDTO;
import com.ard.tianyu.websocket.utils.WebSocketUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@@ -7,17 +13,42 @@
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 UavControlManager uavControlManager;
+ @Resource
+ private UavStreamManager uavStreamManager;
@Override
public void handleTextMessage(WebSocketSession session, TextMessage message) {
// 处理接收到的消息
+ String userId = (String) session.getAttributes().get("userId");
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":
+ BaseMessage<PullStreamDTO> streamMessage = JSON.parseObject(payload,
+ new TypeReference<BaseMessage<PullStreamDTO>>() {
+ });
+ PullStreamDTO pullStreamDTO = streamMessage.getPayload();
+ if (pullStreamDTO != null) {
+ String boxSn = pullStreamDTO.getBoxSn();
+ if (pullStreamDTO.getPullType().equals(1)) {
+ uavStreamManager.startPullStream(boxSn, userId);
+ } else {
+ uavStreamManager.stopPullStream(boxSn, userId);
+ }
+ }
+ break;
+ }
+
}
@Override
@@ -35,5 +66,9 @@
String userId = (String) session.getAttributes().get("userId");
WebSocketUtil.ONLINE_USER_SESSIONS.remove(userId);
log.info("【cmd】size: {}", WebSocketUtil.ONLINE_USER_SESSIONS.size());
+ //停止用户的所有流
+ uavStreamManager.stopAllPullStreamByUser(userId);
+ //移除控制的无人机
+ uavControlManager.releaseAllControlByUser(userId);
}
}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java b/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java
deleted file mode 100644
index e743676..0000000
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java
+++ /dev/null
@@ -1,123 +0,0 @@
-package com.ard.tianyu.websocket.handler;
-
-import com.alibaba.fastjson2.JSON;
-import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver;
-import com.ard.tianyu.service.ITyService;
-import com.ard.tianyu.websocket.model.PullStreamDTO;
-import com.ard.tianyu.websocket.utils.WebSocketUtil;
-import com.ard.vtdu.domain.ArdVtdu;
-import com.ard.vtdu.service.IMediaService;
-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.WebSocketMessage;
-import org.springframework.web.socket.WebSocketSession;
-import org.springframework.web.socket.handler.TextWebSocketHandler;
-
-import javax.annotation.Resource;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-
-@Component
-@Slf4j
-public class TyStreamHandler extends TextWebSocketHandler {
-
- @Resource
- private ITyService tyService;
- @Resource
- private IMediaService mediaService;
-
- //盒子拉流地址缓存
- private static final ConcurrentMap<String, String> BOX_STREAM_PULL_MAP = new ConcurrentHashMap<>();
- //盒子拉流用户集合缓存
- private static final ConcurrentMap<String, Set<String>> STREAM_USER_MAP = new ConcurrentHashMap<>();
-
- @Override
- public void handleTextMessage(WebSocketSession session, TextMessage message) {
- String userId = (String) session.getAttributes().get("userId");
- String payload = message.getPayload();
- PullStreamDTO pullStreamDTO = JSON.parseObject(payload, PullStreamDTO.class);
-
- if (pullStreamDTO != null) {
- String boxSn = pullStreamDTO.getBoxSn();
- if (pullStreamDTO.getPullType().equals(1)) {
- startPullStream(boxSn, userId);
- } else {
- stopPullStream(boxSn, userId);
- }
- }
-
- log.info("【stream】Received message: {}", payload);
- log.info("【stream】size: {}", WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
- }
-
- //开始拉流
- private void startPullStream(String boxSn, String userId) {
- STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
- String messageReply;
- if (BOX_STREAM_PULL_MAP.containsKey(boxSn)) {
- String webrtcUrl = BOX_STREAM_PULL_MAP.get(boxSn);
- messageReply = webrtcUrl;
- log.warn("【stream】Already pulled stream: {}", webrtcUrl);
- } else {
- BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
- if (boxLiveUrl != null) {
- ArdVtdu vtdu = ArdVtdu.builder().name(boxSn).mode(0).isCode(0).rtspSource(boxLiveUrl.getRtsp()).build();
- ArdVtdu ardVtdu = mediaService.addPath(vtdu);
- BOX_STREAM_PULL_MAP.put(boxSn, ardVtdu.getWebrtcUrl());
- messageReply = ardVtdu.getWebrtcUrl();
- log.warn("【stream】First pulled stream: {}", ardVtdu.getWebrtcUrl());
- } else {
- log.warn("【stream】Box stream not found: {}", boxSn);
- messageReply = "【stream】Box stream not found: " + boxSn;
- }
- }
- WebSocketUtil.sendMessage(WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.get(userId), messageReply);
- }
-
- //停止拉流
- private void stopPullStream(String boxSn, String userId) {
- synchronized (this) {
- Set<String> users = STREAM_USER_MAP.get(boxSn);
- if (users != null) {
- users.remove(userId);
- int userCount = users.size();
- String messageReply;
- if (userCount == 0) {
- closeStream(boxSn);
- messageReply = "【stream】Stream closed due to no active users: " + boxSn;
- } else {
- messageReply = "【stream】Stream is still in use, remaining users: " + users;
- }
- WebSocketUtil.sendMessage(WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.get(userId), messageReply);
- }
- }
- }
-
- //关闭拉流
- private void closeStream(String boxSn) {
- mediaService.removePath(boxSn);
- BOX_STREAM_PULL_MAP.remove(boxSn);
- STREAM_USER_MAP.remove(boxSn);
- log.warn("【stream】Stream closed due to no active users: {}", boxSn);
- }
-
- @Override
- public void afterConnectionEstablished(WebSocketSession session) {
- String userId = (String) session.getAttributes().get("userId");
- WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.put(userId, session);
- log.info("【stream】Connected: {}", session.getId());
- log.info("【stream】size: {}", WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
- }
-
- @Override
- public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
- String userId = (String) session.getAttributes().get("userId");
- WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.remove(userId);
- STREAM_USER_MAP.forEach((boxSn, users) -> stopPullStream(boxSn, userId));
- log.info("【stream】Disconnected: {}", session.getId());
- log.info("【stream】size: {}", WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
- }
-}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java b/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java
index 7090119..27c8e65 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java
@@ -27,9 +27,6 @@
// 存储 websocket session
public static final ConcurrentMap<String, WebSocketSession> ONLINE_USER_SESSIONS = new ConcurrentHashMap<>();
- // 存储 拉流 websocket session
- public static final ConcurrentMap<String, WebSocketSession> STREAM_ONLINE_USER_SESSIONS = new ConcurrentHashMap<>();
-
/**
* @param session 用户 session
* @param message 发送内容
@@ -51,10 +48,12 @@
}
}
}
+
public static void sendMessage(String userId, String message) {
WebSocketSession session = ONLINE_USER_SESSIONS.get(userId);
sendMessage(session, message);
}
+
/**
* @param session 用户 session
* @param message 发送内容
@@ -63,7 +62,7 @@
if (session == null) {
return;
}
- final InetSocketAddress remoteAddress = session.getRemoteAddress();
+ final InetSocketAddress remoteAddress = session.getRemoteAddress();
if (remoteAddress == null) {
return;
}
@@ -101,6 +100,7 @@
public static void sendMessageAll(String message) {
ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, message));
}
+
/**
* 推送消息到其他客户端
*
@@ -110,6 +110,7 @@
JSONObject jsonObject = new JSONObject(message);
ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, jsonObject.toString()));
}
+
/**
* 推送消息到其他客户端
*
@@ -119,6 +120,7 @@
String jsonMessage = JSONObject.toJSONString(message); // 确保对象转换为 JSON 字符串
ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, jsonMessage));
}
+
/**
* 推送消息到其他客户端
*
@@ -127,6 +129,7 @@
public static void sendMessageAll(List message) {
ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, JSON.toJSONString(message)));
}
+
/**
* 发送消息给userId_前缀的人
*/
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WsMessageConverter.java b/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WsMessageConverter.java
deleted file mode 100644
index ba61f18..0000000
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WsMessageConverter.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package com.ard.tianyu.websocket.utils;
-
-import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.TypeReference;
-import com.ard.tianyu.websocket.domain.WsMessage;
-
-public class WsMessageConverter {
-
- // 将 JSON 字符串转换为 WsMessage 对象
- public static <T> WsMessage<T> fromJson(String json, Class<T> payloadType) {
- // 使用 TypeReference 来处理泛型
- return JSON.parseObject(json, new TypeReference<WsMessage<T>>(){});
- }
-
- // 将 WsMessage 对象转换为 JSON 字符串
- public static <T> String toJson(WsMessage<T> wsMessage) {
- return JSON.toJSONString(wsMessage);
- }
-}
--
Gitblit v1.9.3