From c8c7f352db0b031238a5840ce6536e50e43327c6 Mon Sep 17 00:00:00 2001 From: tongtongstudio Date: Wed, 15 Jul 2026 14:53:20 +0800 Subject: [PATCH] build: change file name --- pom.xml | 6 +- ...n.java => WebRtcSignalingApplication.java} | 4 +- .../service/WebRTCSignalingServer.java | 218 +++++++----------- src/main/resources/application.properties | 2 +- ...a => WebRtcSignalingApplicationTests.java} | 2 +- 5 files changed, 93 insertions(+), 139 deletions(-) rename src/main/java/com/ttstd/signaling/{OneKeyCallWebRtcSignalingApplication.java => WebRtcSignalingApplication.java} (63%) rename src/test/java/com/ttstd/signaling/{OneKeyCallWebRtcSignalingApplicationTests.java => WebRtcSignalingApplicationTests.java} (83%) diff --git a/pom.xml b/pom.xml index 5e96709..e875b7e 100644 --- a/pom.xml +++ b/pom.xml @@ -9,10 +9,10 @@ com.ttstd - OneKeyCallWebRTCSignaling + WebRTCSignaling 0.0.1-SNAPSHOT - OneKeyCallWebRTCSignaling - OneKeyCallWebRTCSignaling + WebRTCSignaling + WebRTCSignaling diff --git a/src/main/java/com/ttstd/signaling/OneKeyCallWebRtcSignalingApplication.java b/src/main/java/com/ttstd/signaling/WebRtcSignalingApplication.java similarity index 63% rename from src/main/java/com/ttstd/signaling/OneKeyCallWebRtcSignalingApplication.java rename to src/main/java/com/ttstd/signaling/WebRtcSignalingApplication.java index 39efae3..10ce238 100644 --- a/src/main/java/com/ttstd/signaling/OneKeyCallWebRtcSignalingApplication.java +++ b/src/main/java/com/ttstd/signaling/WebRtcSignalingApplication.java @@ -4,10 +4,10 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication -public class OneKeyCallWebRtcSignalingApplication { +public class WebRtcSignalingApplication { public static void main(String[] args) { - SpringApplication.run(OneKeyCallWebRtcSignalingApplication.class, args); + SpringApplication.run(WebRtcSignalingApplication.class, args); } } diff --git a/src/main/java/com/ttstd/signaling/service/WebRTCSignalingServer.java b/src/main/java/com/ttstd/signaling/service/WebRTCSignalingServer.java index e30a5ba..1504d02 100644 --- a/src/main/java/com/ttstd/signaling/service/WebRTCSignalingServer.java +++ b/src/main/java/com/ttstd/signaling/service/WebRTCSignalingServer.java @@ -15,19 +15,35 @@ import org.springframework.stereotype.Component; import java.io.IOException; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +/** + * WebRTC 信令中继服务。 + * + *

职责：作为 Client（观看端）与 Service（推流端）之间的无状态消息中转站。 + * 收到消息后解析出目标用户，在线则直接转发，离线则交由离线缓冲（Redis）暂存， + * 待对端上线后由 {@link #onOpen} 补发，从而保证信令不丢失、不出现死锁。

+ * + *

协议：外层信封 {@link WebRTCMessage}{type, target, data}， + * 内层载荷 {@link SignalingMessage}{type, senderId, targetId, payload}。

+ */ @Component @ServerEndpoint("/signaling/{userId}") public class WebRTCSignalingServer { - Logger logger = LoggerFactory.getLogger(WebRTCSignalingServer.class); + private static final Logger logger = LoggerFactory.getLogger(WebRTCSignalingServer.class); - // 用于存储在线用户的 Session + /** 在线用户 Session 表 */ private static final Map clients = new ConcurrentHashMap<>(); private static final ObjectMapper objectMapper = new ObjectMapper(); - // RedisTemplate for storing and retrieving messages for offline users + /** 需要中转的信令类型白名单（其余类型由服务器内部处理或不转发） */ + private static final Set RELAY_TYPES = Set.of( + "init", "ask", "offer", "answer", "accept", "reject", + "sdp", "iceCandidate", "call", "hangup" + ); + private static RedisTemplate redisTemplate; @Autowired @@ -38,15 +54,14 @@ public class WebRTCSignalingServer { @OnOpen public void onOpen(Session session, @PathParam("userId") String userId) { clients.put(userId, session); - logger.info("onOpen: 用户连接: " + userId); + logger.info("onOpen: 用户连接: {}", userId); - // Check for pending messages in Redis for the newly connected user + // 用户上线后补发其离线期间暂存的消息 String redisKey = "webrtc:pending_messages:" + userId; if (redisTemplate != null) { Long size = redisTemplate.opsForList().size(redisKey); if (size != null && size > 0) { logger.info("onOpen: 用户 {} 上线，发现 {} 条待发送消息。", userId, size); - // Retrieve and send all pending messages for (int i = 0; i < size; i++) { WebRTCMessage pendingMessage = (WebRTCMessage) redisTemplate.opsForList().leftPop(redisKey); if (pendingMessage != null) { @@ -57,7 +72,7 @@ public class WebRTCSignalingServer { logger.error("onOpen: 序列化待发送消息失败: {}", e.getMessage()); } catch (IOException e) { logger.error("onOpen: 发送待发送消息给用户 {} 失败: {}", userId, e.getMessage()); - // If sending fails (e.g., network issue), push it back to Redis + // 发送失败（如网络抖动）重新入队，等待下次补发 redisTemplate.opsForList().rightPush(redisKey, pendingMessage); } } @@ -74,47 +89,27 @@ public class WebRTCSignalingServer { try { WebRTCMessage webRTCMessage = objectMapper.readValue(message, WebRTCMessage.class); - if (webRTCMessage == null) { - logger.warn("onMessage: 无法解析消息，请检查消息格式。"); + if (webRTCMessage == null || webRTCMessage.getType() == null) { + logger.warn("onMessage: 无法解析消息或缺少 type 字段。"); return; } String type = webRTCMessage.getType(); switch (type) { - case "init": - - break; - case "ask": - // Handle ask message - case "offer": - // Handle offer message - case "answer": - // Handle answer messages - case "reject": - // Handle reject message - case "accept": - // Handle accept message - handleWebRTCMessage(session, webRTCMessage, userId); - break; - case "sdp": - // Handle SessionDescription messages - case "iceCandidate": - // Handle iceCandidate messages - handleSdpMessage(session, webRTCMessage, userId); - break; - - case "call": - // Handle call message - break; - case "hangup": - // Handle hangup message - break; case "ping": - // Handle ping message + // 应用层心跳保活，无需转发 + logger.debug("onMessage: 收到来自 {} 的心跳，忽略。", userId); break; case "unknown": + case "error": + logger.warn("onMessage: 收到 {} 消息，不进行处理。", type); + break; default: - // Handle unknown message type + if (RELAY_TYPES.contains(type)) { + relayMessage(session, webRTCMessage, userId); + } else { + logger.warn("onMessage: 未知消息类型 {}，已忽略。", type); + } break; } } catch (JsonProcessingException e) { @@ -126,103 +121,66 @@ public class WebRTCSignalingServer { } } - private void handleWebRTCMessage(Session session, WebRTCMessage webRTCMessage, String userId) throws JsonProcessingException { - logger.info("handleWebRTCMessage: 收到来自用户 {} 的消息: {}", userId, webRTCMessage); - String type = webRTCMessage.getType(); - - SignalingMessage signalingMessage; - if (webRTCMessage.getData() instanceof String) { - signalingMessage = objectMapper.readValue((String) webRTCMessage.getData(), SignalingMessage.class); - } else { - signalingMessage = objectMapper.convertValue(webRTCMessage.getData(), SignalingMessage.class); + /** + * 统一中继：将信令转发给目标用户。 + *
    + *
  • 目标在线 → 直接异步发送；
    • + *
  • 目标离线且有 Redis → 暂存，待其上线补发（避免状态丢失）；
    • + *
  • 目标离线且无 Redis → 向发送方返回错误。
    • + *
+ */ + private void relayMessage(Session session, WebRTCMessage webRTCMessage, String fromUserId) { + String targetId = extractTargetId(webRTCMessage); + if (targetId == null || targetId.isEmpty()) { + logger.warn("relayMessage: 缺少目标用户ID，无法转发。消息类型: {}", webRTCMessage.getType()); + sendErrorMessage(session, fromUserId, "消息格式错误，缺少目标用户ID。"); + return; } - String targetId = signalingMessage.getTargetId(); - - if (targetId != null && !targetId.isEmpty()) { - if (clients.containsKey(targetId)) { - // Target user is online, send message directly - Session targetSession = clients.get(targetId); - if (targetSession != null && targetSession.isOpen()) { -// targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage)); - logger.info("onMessage: 消息从 {} 发送给在线用户 {}", userId, targetId); - switch (type) { - case "ask":// Handle ask message - targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage)); - break; - case "offer":// Handle offer message - targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage)); - break; - case "answer":// Handle answer messages - - break; - case "reject":// Handle reject message - - break; - case "accept":// Handle accept message - - break; - } - } else { - clients.remove(targetId); - logger.warn("onMessage: 目标用户 {} 的会话已关闭", targetId); - } - } else { - sendErrorMessage(session, userId, "onMessage: 目标用户 {" + targetId + "} 不在线"); - logger.info("onMessage: 目标用户 {} 不在线", targetId); + Session targetSession = clients.get(targetId); + if (targetSession != null && targetSession.isOpen()) { + try { + targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage)); + logger.info("relayMessage: 消息 {} 已从 {} 转发给在线用户 {}", webRTCMessage.getType(), fromUserId, targetId); + } catch (JsonProcessingException e) { + logger.error("relayMessage: 序列化消息失败: {}", e.getMessage()); + sendErrorMessage(session, fromUserId, "消息序列化失败。"); } } else { - logger.warn("onMessage: 收到消息但缺少 targetId 或 targetId 为空，无法转发或存储。消息内容: {}", webRTCMessage); - sendErrorMessage(session, userId, "消息格式错误，缺少目标用户ID。"); + // 目标离线：优先离线缓冲，保证信令不丢 + if (redisTemplate != null) { + storeMessageInRedis(targetId, webRTCMessage); + logger.info("relayMessage: 目标 {} 离线，消息 {} 已暂存，待其上线补发。", targetId, webRTCMessage.getType()); + } else { + clients.remove(targetId); + logger.info("relayMessage: 目标用户 {} 不在线（无 Redis 缓冲）", targetId); + sendErrorMessage(session, fromUserId, "目标用户 {" + targetId + "} 不在线"); + } } } - private void handleSdpMessage(Session session, WebRTCMessage webRTCMessage, String userId) throws JsonProcessingException { - logger.info("handleSdpMessage: 收到来自用户 {} 的消息: {}", userId, webRTCMessage); - String type = webRTCMessage.getType(); - - SignalingMessage signalingMessage; - if (webRTCMessage.getData() instanceof String) { - signalingMessage = objectMapper.readValue((String) webRTCMessage.getData(), SignalingMessage.class); - } else { - signalingMessage = objectMapper.convertValue(webRTCMessage.getData(), SignalingMessage.class); - } - - String targetId = signalingMessage.getTargetId(); - - if (targetId != null && !targetId.isEmpty()) { - if (clients.containsKey(targetId)) { - // Target user is online, send message directly - Session targetSession = clients.get(targetId); - if (targetSession != null && targetSession.isOpen()) { -// targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage)); - logger.info("onMessage: 消息从 {} 发送给在线用户 {}", userId, targetId); - switch (type) { - case "sdp":// Handle SessionDescription message - targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage)); - break; - case "iceCandidate":// Handle iceCandidate message - targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage)); - break; - } - } else { - clients.remove(targetId); - logger.warn("onMessage: 目标用户 {} 的会话已关闭", targetId); + /** 从消息中解析目标用户ID：优先取内层 SignalingMessage.targetId，回退到外层 target */ + private String extractTargetId(WebRTCMessage message) { + Object data = message.getData(); + if (data != null) { + try { + SignalingMessage sm = (data instanceof String) + ? objectMapper.readValue((String) data, SignalingMessage.class) + : objectMapper.convertValue(data, SignalingMessage.class); + if (sm != null && sm.getTargetId() != null && !sm.getTargetId().isEmpty()) { + return sm.getTargetId(); } - } else { - sendErrorMessage(session, userId, "onMessage: 目标用户 {" + targetId + "} 不在线"); - logger.info("onMessage: 目标用户 {} 不在线", targetId); + } catch (Exception ignored) { + // 解析失败时回退到外层 target } - } else { - logger.warn("onMessage: 收到消息但缺少 targetId 或 targetId 为空，无法转发或存储。消息内容: {}", webRTCMessage); - sendErrorMessage(session, userId, "消息格式错误，缺少目标用户ID。"); } + return (message.getTarget() != null && !message.getTarget().isEmpty()) ? message.getTarget() : null; } @OnClose public void onClose(@PathParam("userId") String userId) { clients.remove(userId); - logger.info("onClose: 用户离开: " + userId); + logger.info("onClose: 用户离开: {}", userId); } @OnError @@ -235,7 +193,7 @@ public class WebRTCSignalingServer { WebRTCMessage errorMessage = new WebRTCMessage(); errorMessage.setType("error"); errorMessage.setTarget(userId); - errorMessage.setData(errorMsg); // Include the error message in data + errorMessage.setData(errorMsg); session.getAsyncRemote().sendText(objectMapper.writeValueAsString(errorMessage)); logger.info("sendErrorMessage: 发送错误消息: {}", errorMsg); } catch (JsonProcessingException e) { @@ -244,16 +202,12 @@ public class WebRTCSignalingServer { } private void storeMessageInRedis(String targetId, WebRTCMessage message) { - if (redisTemplate != null) { - try { - String redisKey = "webrtc:pending_messages:" + targetId; - redisTemplate.opsForList().rightPush(redisKey, message); - logger.info("storeMessageInRedis: 消息已存储到 Redis，key: {}", redisKey); - } catch (Exception e) { - logger.error("storeMessageInRedis: 存储消息到 Redis 失败: {}", e.getMessage(), e); - } - } else { - logger.warn("RedisTemplate is not initialized. Cannot store message for user {}.", targetId); + try { + String redisKey = "webrtc:pending_messages:" + targetId; + redisTemplate.opsForList().rightPush(redisKey, message); + logger.info("storeMessageInRedis: 消息已存储到 Redis，key: {}", redisKey); + } catch (Exception e) { + logger.error("storeMessageInRedis: 存储消息到 Redis 失败: {}", e.getMessage(), e); } } } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index e7f0179..681f95e 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -1,4 +1,4 @@ -spring.application.name=OneKeyCallWebRTCSignaling +spring.application.name=WebRTCSignaling server.port=2310 spring.data.redis.database=5 diff --git a/src/test/java/com/ttstd/signaling/OneKeyCallWebRtcSignalingApplicationTests.java b/src/test/java/com/ttstd/signaling/WebRtcSignalingApplicationTests.java similarity index 83% rename from src/test/java/com/ttstd/signaling/OneKeyCallWebRtcSignalingApplicationTests.java rename to src/test/java/com/ttstd/signaling/WebRtcSignalingApplicationTests.java index c5e7fe3..eacf3cc 100644 --- a/src/test/java/com/ttstd/signaling/OneKeyCallWebRtcSignalingApplicationTests.java +++ b/src/test/java/com/ttstd/signaling/WebRtcSignalingApplicationTests.java @@ -4,7 +4,7 @@ import org.junit.jupiter.api.Test; import org.springframework.boot.test.context.SpringBootTest; @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) -class OneKeyCallWebRtcSignalingApplicationTests { +class WebRtcSignalingApplicationTests { @Test void contextLoads() {