tongtong/OneKeyCallWebRTCSignaling
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

build: change file name

Browse Source
This commit is contained in:
tongtongstudio
2026-07-15 14:53:20 +08:00
parent 4ca1c01aea
commit c8c7f352db
5 changed files with 93 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pom.xml
View File

@@ -9,10 +9,10 @@
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.ttstd</groupId>
<artifactId>OneKeyCallWebRTCSignaling</artifactId>
<artifactId>WebRTCSignaling</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>OneKeyCallWebRTCSignaling</name>
<description>OneKeyCallWebRTCSignaling</description>
<name>WebRTCSignaling</name>
<description>WebRTCSignaling</description>
<url/>
<licenses>
<license/>

4
src/main/java/com/ttstd/signaling/OneKeyCallWebRtcSignalingApplication.java → src/main/java/com/ttstd/signaling/WebRtcSignalingApplication.java
View File

@@ -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);
}
}

218
src/main/java/com/ttstd/signaling/service/WebRTCSignalingServer.java
View File

@@ -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 信令中继服务。
*
* <p>职责：作为 Client（观看端）与 Service（推流端）之间的无状态消息中转站。
* 收到消息后解析出目标用户，在线则直接转发，离线则交由离线缓冲（Redis）暂存，
* 待对端上线后由 {@link #onOpen} 补发，从而保证信令不丢失、不出现死锁。</p>
*
* <p>协议：外层信封 {@link WebRTCMessage}{type, target, data}，
* 内层载荷 {@link SignalingMessage}{type, senderId, targetId, payload}。</p>
*/
@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<String, Session> clients = new ConcurrentHashMap<>();
private static final ObjectMapper objectMapper = new ObjectMapper();
// RedisTemplate for storing and retrieving messages for offline users
/** 需要中转的信令类型白名单（其余类型由服务器内部处理或不转发） */
private static final Set<String> RELAY_TYPES = Set.of(
"init", "ask", "offer", "answer", "accept", "reject",
"sdp", "iceCandidate", "call", "hangup"
);
private static RedisTemplate<String, Object> 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);
/**
* 统一中继：将信令转发给目标用户。
* <ul>
* <li>目标在线 → 直接异步发送；</li>
* <li>目标离线且有 Redis → 暂存，待其上线补发（避免状态丢失）；</li>
* <li>目标离线且无 Redis → 向发送方返回错误。</li>
* </ul>
*/
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);
}
}
}

2
src/main/resources/application.properties
View File

@@ -1,4 +1,4 @@
spring.application.name=OneKeyCallWebRTCSignaling
spring.application.name=WebRTCSignaling
server.port=2310
spring.data.redis.database=5

2
src/test/java/com/ttstd/signaling/OneKeyCallWebRtcSignalingApplicationTests.java → src/test/java/com/ttstd/signaling/WebRtcSignalingApplicationTests.java
View File

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