feat: 信令交换，还不能显示

pom.xml

@@ -57,6 +57,19 @@
             <artifactId>gson</artifactId>
             <version>2.10.1</version>
         </dependency>
+
+        <dependency>
+            <groupId>dev.onvoid.webrtc</groupId>
+            <artifactId>webrtc-java</artifactId>
+            <version>0.8.0</version>
+        </dependency>
+
+        <!-- Redis dependency -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
+
     </dependencies>
 
     <build>

src/main/java/com/ttstd/signaling/bean/SignalingMessage.java

@@ -0,0 +1,53 @@
+package com.ttstd.signaling.bean;
+
+import java.io.Serializable;
+
+public class SignalingMessage implements Serializable {
+    private String type;       // "call", "offer", "answer", "iceCandidate", etc.
+    private String senderId;   // 发送者ID
+    private String targetId;   // 接收者ID
+    private Object payload;    // 具体携带的数据（可以是String, SessionDescription, IceCandidate等）
+
+    // Default constructor for JSON deserialization
+    public SignalingMessage() {
+    }
+
+    public SignalingMessage(String type, String senderId, String targetId, Object payload) {
+        this.type = type;
+        this.senderId = senderId;
+        this.targetId = targetId;
+        this.payload = payload;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
+    public String getSenderId() {
+        return senderId;
+    }
+
+    public void setSenderId(String senderId) {
+        this.senderId = senderId;
+    }
+
+    public String getTargetId() {
+        return targetId;
+    }
+
+    public void setTargetId(String targetId) {
+        this.targetId = targetId;
+    }
+
+    public Object getPayload() {
+        return payload;
+    }
+
+    public void setPayload(Object payload) {
+        this.payload = payload;
+    }
+}

src/main/java/com/ttstd/signaling/config/RedisConfig.java

@@ -0,0 +1,34 @@
+package com.ttstd.signaling.config;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+@Configuration
+public class RedisConfig {
+
+    @Bean
+    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
+        RedisTemplate<String, Object> template = new RedisTemplate<>();
+        template.setConnectionFactory(redisConnectionFactory);
+
+        // String serializer for keys
+        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
+        template.setKeySerializer(stringRedisSerializer);
+        template.setHashKeySerializer(stringRedisSerializer);
+
+        // JSON serializer for values
+        GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();
+        template.setValueSerializer(genericJackson2JsonRedisSerializer);
+        template.setHashValueSerializer(genericJackson2JsonRedisSerializer);
+
+        template.afterPropertiesSet();
+        return template;
+    }
+}

src/main/java/com/ttstd/signaling/model/WebRTCMessage.java

@@ -0,0 +1,10 @@
+package com.ttstd.signaling.model;
+
+import lombok.Data;
+
+@Data
+public class WebRTCMessage {
+    private String type; // "offer", "answer", "iceCandidate", "call", "hangup", "reject", "accept"
+    private String target;
+    private Object data;
+}

src/main/java/com/ttstd/signaling/service/WebRTCSignalingServer.java

@@ -1,24 +1,19 @@
 package com.ttstd.signaling.service;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.ttstd.signaling.bean.SignalingMessage;
-import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.ttstd.signaling.model.WebRTCMessage;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.google.gson.Gson;
-import com.google.gson.JsonObject;
-import com.ttstd.signaling.bean.IceCandidate;
-import com.ttstd.signaling.bean.SessionDescription;
-import com.ttstd.signaling.model.*;
 import jakarta.websocket.*;
 import jakarta.websocket.server.PathParam;
 import jakarta.websocket.server.ServerEndpoint;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
 
-import java.util.HashMap;
+import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -28,158 +23,200 @@ public class WebRTCSignalingServer {
 
     Logger logger = LoggerFactory.getLogger(WebRTCSignalingServer.class);
 
-    private Gson gson = new Gson();
     // 用于存储在线用户的 Session
     private static final Map<String, Session> clients = new ConcurrentHashMap<>();
     private static final ObjectMapper objectMapper = new ObjectMapper();
 
-    private static final Map<String, SessionDescription> serviceSessionDescriptionMap = new ConcurrentHashMap<>();
+    // RedisTemplate for storing and retrieving messages for offline users
-    private static final Map<String, IceCandidate> serviceIceCandidateMap = new ConcurrentHashMap<>();
+    private static RedisTemplate<String, Object> redisTemplate;
 
-    private static final Map<String, SessionDescription> clientSessionDescriptionMap = new ConcurrentHashMap<>();
+    @Autowired
-    private static final Map<String, IceCandidate> clientIceCandidateMap = new ConcurrentHashMap<>();
+    public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
+        WebRTCSignalingServer.redisTemplate = redisTemplate;
+    }
 
     @OnOpen
     public void onOpen(Session session, @PathParam("userId") String userId) {
         clients.put(userId, session);
         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) {
+                        try {
+                            session.getAsyncRemote().sendText(objectMapper.writeValueAsString(pendingMessage));
+                            logger.info("onOpen: 已将待发送消息发送给用户 {}: {}", userId, pendingMessage.getType());
+                        } catch (JsonProcessingException e) {
+                            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);
+                        }
+                    }
+                }
+            }
+        } else {
+            logger.warn("RedisTemplate is not initialized. Cannot check for pending messages.");
+        }
     }
 
     @OnMessage
     public void onMessage(String message, Session session, @PathParam("userId") String userId) {
-        logger.info("onMessage: message = " + message);
+        logger.info("onMessage: 收到来自用户 {} 的消息: {}", userId, message);
 
         try {
-            JsonNode jsonNode = objectMapper.readTree(message);
+            WebRTCMessage webRTCMessage = objectMapper.readValue(message, WebRTCMessage.class);
-            int msgType = jsonNode.get("msg_type").asInt();
+            if (webRTCMessage == null) {
-            logger.info("onMessage: msgType = " + msgType);
+                logger.warn("onMessage: 无法解析消息，请检查消息格式。");
-            String content = jsonNode.get("content").toString();
+                return;
-            logger.info("onMessage: content = " + content);
-            switch (msgType) {
-                case 1:
-                    receiverServiceSdp(content);
-                    break;
-                case 2:
-                    receiverServiceIce(content);
-                    break;
-                case 3:
-                    receiverClientOfferMessage(content);
-                    break;
-                case 4:
-                    receiverClientSdpMessage(content);
-                    break;
-                case 5:
-                    receiverClientIceMessage(content);
-                    break;
-                case -99:
-                    // 解析收到的 JSON 消息
-                    SignalMessage signal = objectMapper.readValue(message, SignalMessage.class);
-                    String toUser = signal.getToUser();
-                    Session targetSession = clients.get(toUser);
-
-                    if (targetSession != null && targetSession.isOpen()) {
-                        // 转发信令（SDP, ICE, 拒绝/接受）
-                        targetSession.getAsyncRemote().sendText(message);
-                        logger.info("onMessage: 信令类型 [" + signal.getType() + "] 从 " + userId + " 发往 " + toUser);
-                    } else {
-                        sendErrorMessage(session, "目标用户不在线: " + toUser);
-                    }
-                    break;
             }
 
+            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
+                    break;
+                case "unknown":
+                default:
+                    // Handle unknown message type
+                    break;
+            }
+        } catch (JsonProcessingException e) {
+            logger.error("onMessage: JSON 解析失败: {}", e.getMessage());
+            sendErrorMessage(session, userId, "消息格式错误，无法解析。");
         } catch (Exception e) {
-            e.printStackTrace();
+            logger.error("onMessage: 处理消息时发生未知错误: {}", e.getMessage(), e);
+            sendErrorMessage(session, userId, "服务器内部错误。");
         }
     }
 
-    /**
+    private void handleWebRTCMessage(Session session, WebRTCMessage webRTCMessage, String userId) throws JsonProcessingException {
-     * 接收并存储被控端sdp信息
+        logger.info("handleWebRTCMessage: 收到来自用户 {} 的消息: {}", userId, webRTCMessage);
-     * @param jsonString
+        String type = webRTCMessage.getType();
-     * @throws JsonProcessingException
+
-     */
+        SignalingMessage signalingMessage;
-    private void receiverServiceSdp(String jsonString) throws JsonProcessingException {
+        if (webRTCMessage.getData() instanceof String) {
-        // 解析收到的 JSON 消息
+            signalingMessage = objectMapper.readValue((String) webRTCMessage.getData(), SignalingMessage.class);
-        ServiceSdpMessage serviceSdpMessage = objectMapper.readValue(jsonString, ServiceSdpMessage.class);
+        } else {
-        String fromUser = serviceSdpMessage.getFromUser();
+            signalingMessage = objectMapper.convertValue(webRTCMessage.getData(), SignalingMessage.class);
-        Session targetSession = clients.get(fromUser);
-        logger.info("receiverServiceSdp: fromUser = " + fromUser);
-        logger.info("receiverServiceSdp: targetSession = " + targetSession);
-        SessionDescription sessionDescription = new SessionDescription(serviceSdpMessage.getType(), serviceSdpMessage.getSdp());
-        serviceSessionDescriptionMap.put(fromUser, sessionDescription);
-        logger.info("receiverServiceSdp: serviceSessionDescriptionMap size = " + serviceSessionDescriptionMap.size());
         }
 
-    /**
+        String targetId = signalingMessage.getTargetId();
-     * 接收并存储被控端ice信息
+
-     * @param jsonString
+        if (targetId != null && !targetId.isEmpty()) {
-     * @throws JsonProcessingException
+            if (clients.containsKey(targetId)) {
-     */
+                // Target user is online, send message directly
-    private void receiverServiceIce(String jsonString) throws JsonProcessingException {
+                Session targetSession = clients.get(targetId);
-        ServiceIceMessage serviceIceMessage = objectMapper.readValue(jsonString, ServiceIceMessage.class);
+                if (targetSession != null && targetSession.isOpen()) {
-        String fromUser = serviceIceMessage.getFromUser();
+//                    targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage));
-        Session targetSession = clients.get(fromUser);
+                    logger.info("onMessage: 消息从 {} 发送给在线用户 {}", userId, targetId);
-        logger.info("receiverServiceIce: fromUser = " + fromUser);
+                    switch (type) {
-        logger.info("receiverServiceIce: targetSession = " + targetSession);
+                        case "ask":// Handle ask message
-        IceCandidate iceCandidate = new IceCandidate(serviceIceMessage.getSdpMid(), serviceIceMessage.getSdpMLineIndex(), serviceIceMessage.getSdp());
+                            targetSession.getAsyncRemote().sendText(objectMapper.writeValueAsString(webRTCMessage));
-        serviceIceCandidateMap.put(fromUser, iceCandidate);
+                            break;
-        logger.info("receiverServiceIce: serviceIceCandidateMap size = " + serviceIceCandidateMap.size());
+                        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);
+            }
+        } else {
+            logger.warn("onMessage: 收到消息但缺少 targetId 或 targetId 为空，无法转发或存储。消息内容: {}", webRTCMessage);
+            sendErrorMessage(session, userId, "消息格式错误，缺少目标用户ID。");
+        }
     }
 
-    /**
+    private void handleSdpMessage(Session session, WebRTCMessage webRTCMessage, String userId) throws JsonProcessingException {
-     * 接收并存储控制端sdp信息
+        logger.info("handleSdpMessage: 收到来自用户 {} 的消息: {}", userId, webRTCMessage);
-     * @param jsonString
+        String type = webRTCMessage.getType();
-     * @throws JsonProcessingException
-     */
-    private void receiverClientOfferMessage(String jsonString) throws JsonProcessingException {
-        ClientOfferMessage clientOfferMessage = objectMapper.readValue(jsonString, ClientOfferMessage.class);
-        String fromUser = clientOfferMessage.getFromUser();
-        logger.info("receiverClientOfferMessage: fromUser = " + fromUser);
-        String toUser = clientOfferMessage.getToUser();
-        logger.info("receiverClientOfferMessage: toUser = " + toUser);
-
-        logger.info("receiverClientOfferMessage: serviceSessionDescriptionMap size = " + serviceSessionDescriptionMap.size());
-        logger.info("receiverClientOfferMessage: serviceIceCandidateMap size = " + serviceIceCandidateMap.size());
-
-        Session session = clients.get(fromUser);
-
-        SessionDescription sessionDescription = serviceSessionDescriptionMap.get(toUser);
-        JsonObject sdpJsonObject = new JsonObject();
-        sdpJsonObject.addProperty("msg_type", "11");
-        sdpJsonObject.addProperty("sdp", sessionDescription.toString());
-        String sdpJsonString = gson.toJson(sdpJsonObject);
-        logger.info("receiverClientOfferMessage: sdpJsonString = " + sdpJsonString);
-        session.getAsyncRemote().sendText(gson.toJson(sdpJsonObject));
-
-        IceCandidate iceCandidate = serviceIceCandidateMap.get(toUser);
-        JsonObject iceJsonObject = new JsonObject();
-        iceJsonObject.addProperty("msg_type", "22");
-        iceJsonObject.addProperty("ice", iceCandidate.toString());
-        String iceJsonString = gson.toJson(iceJsonObject);
-        logger.info("receiverClientOfferMessage: iceJsonString = " + iceJsonString);
-
-        session.getAsyncRemote().sendText(gson.toJson(iceJsonObject));
 
+        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();
-     * 接收并存储控制端ice信息
-     * @param jsonString
-     * @throws JsonProcessingException
-     */
-    private void receiverClientSdpMessage(String jsonString) throws JsonProcessingException {
-        ClientSdpMessage clientSdpMessage = objectMapper.readValue(jsonString, ClientSdpMessage.class);
-        String fromUser = clientSdpMessage.getFromUser();
-        String toUser = clientSdpMessage.getToUser();
-        SessionDescription description = new SessionDescription(clientSdpMessage.getType(), clientSdpMessage.getSdp());
-        clientSessionDescriptionMap.put(fromUser, description);
-    }
 
-    private void receiverClientIceMessage(String jsonString) throws JsonProcessingException {
+        if (targetId != null && !targetId.isEmpty()) {
-        ClientIceMessage clientIceMessage = objectMapper.readValue(jsonString, ClientIceMessage.class);
+            if (clients.containsKey(targetId)) {
-        String fromUser = clientIceMessage.getFromUser();
+                // Target user is online, send message directly
-        String toUser = clientIceMessage.getToUser();
+                Session targetSession = clients.get(targetId);
-        IceCandidate iceCandidate = new IceCandidate(clientIceMessage.getSdpMid(), clientIceMessage.getSdpMLineIndex(), clientIceMessage.getSdp());
+                if (targetSession != null && targetSession.isOpen()) {
-        clientIceCandidateMap.put(fromUser, iceCandidate);
+//                    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);
+                }
+            } else {
+                sendErrorMessage(session, userId, "onMessage: 目标用户 {" + targetId + "} 不在线");
+                logger.info("onMessage: 目标用户 {} 不在线", targetId);
+            }
+        } else {
+            logger.warn("onMessage: 收到消息但缺少 targetId 或 targetId 为空，无法转发或存储。消息内容: {}", webRTCMessage);
+            sendErrorMessage(session, userId, "消息格式错误，缺少目标用户ID。");
+        }
     }
 
     @OnClose
@@ -190,11 +227,33 @@ public class WebRTCSignalingServer {
 
     @OnError
     public void onError(Session session, Throwable error) {
-        error.printStackTrace();
+        logger.error("onError: 会话 {} 发生错误: {}", session.getId(), error.getMessage(), error);
-        logger.info("onError: " + error.getMessage());
     }
 
-    private void sendErrorMessage(Session session, String errorMsg) {
+    private void sendErrorMessage(Session session, String userId, String errorMsg) {
-        session.getAsyncRemote().sendText("{\"type\":\"error\", \"message\":\"" + errorMsg + "\"}");
+        try {
+            WebRTCMessage errorMessage = new WebRTCMessage();
+            errorMessage.setType("error");
+            errorMessage.setTarget(userId);
+            errorMessage.setData(errorMsg); // Include the error message in data
+            session.getAsyncRemote().sendText(objectMapper.writeValueAsString(errorMessage));
+            logger.info("sendErrorMessage: 发送错误消息: {}", errorMsg);
+        } catch (JsonProcessingException e) {
+            logger.error("sendErrorMessage: 无法序列化错误消息: {}", e.getMessage());
+        }
+    }
+
+    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);
+        }
     }
 }

src/main/resources/application.properties

@@ -1,2 +1,12 @@
 spring.application.name=OneKeyCallWebRTCSignaling
 server.port=2310
+
+spring.data.redis.database=5
+spring.data.redis.host=175.178.213.60
+spring.data.redis.port=26379
+spring.data.redis.password=fanhuitong
+spring.data.redis.timeout=10s
+spring.data.redis.lettuce.pool.max-active=8
+spring.data.redis.lettuce.pool.max-wait=-1
+spring.data.redis.lettuce.pool.max-idle=8
+spring.data.redis.lettuce.pool.min-idle=0