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() {