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

pom.xml

@@ -51,6 +51,25 @@
             <artifactId>lombok</artifactId>
             <optional>true</optional>
         </dependency>
+
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <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/IceCandidate.java

@@ -0,0 +1,53 @@
+package com.ttstd.signaling.bean;
+
+
+import com.google.gson.Gson;
+import com.google.gson.JsonParser;
+
+import java.io.Serializable;
+
+public class IceCandidate implements Serializable {
+
+    String sdpMid;
+    int sdpMLineIndex;
+    String sdp;
+
+    public IceCandidate() {
+    }
+
+    public IceCandidate(String sdpMid, int sdpMLineIndex, String sdp) {
+        this.sdpMid = sdpMid;
+        this.sdpMLineIndex = sdpMLineIndex;
+        this.sdp = sdp;
+    }
+
+    public String getSdpMid() {
+        return sdpMid;
+    }
+
+    public void setSdpMid(String sdpMid) {
+        this.sdpMid = sdpMid;
+    }
+
+    public int getSdpMLineIndex() {
+        return sdpMLineIndex;
+    }
+
+    public void setSdpMLineIndex(int sdpMLineIndex) {
+        this.sdpMLineIndex = sdpMLineIndex;
+    }
+
+    public String getSdp() {
+        return sdp;
+    }
+
+    public void setSdp(String sdp) {
+        this.sdp = sdp;
+    }
+
+    @Override
+    public String toString() {
+        return JsonParser.parseString(new Gson().toJson(this)).getAsJsonObject().toString();
+    }
+}
+

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

@@ -0,0 +1,41 @@
+package com.ttstd.signaling.bean;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonParser;
+
+import java.io.Serializable;
+
+public class SessionDescription implements Serializable {
+    String type;
+    String description;
+
+    public SessionDescription() {
+    }
+
+    public SessionDescription(String type, String description) {
+        this.type = type;
+        this.description = description;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    public void setDescription(String description) {
+        this.description = description;
+    }
+
+    @Override
+    public String toString() {
+        return JsonParser.parseString(new Gson().toJson(this)).getAsJsonObject().toString();
+    }
+
+}

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/controller/AuthController.java

@@ -1,6 +1,5 @@
 package com.ttstd.signaling.controller;
 
-import com.ttstd.signaling.model.SignalingMessage;
 import jakarta.servlet.http.HttpSession;
 import org.springframework.web.bind.annotation.*;
 

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

@@ -0,0 +1,13 @@
+package com.ttstd.signaling.model;
+
+import lombok.Data;
+
+@Data
+public class ClientIceMessage {
+    private String sdpMid;
+    private int sdpMLineIndex;
+    private String sdp;
+    private String fromUser;
+    private String toUser;
+    private long timestamp;
+}

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

@@ -0,0 +1,12 @@
+package com.ttstd.signaling.model;
+
+import lombok.Data;
+
+@Data
+public class ClientOfferMessage {
+    private String type;      // 类型: "offer", "answer", "ice", "reject", "accept"
+    private String fromUser;
+    private String toUser;
+    private long timestamp;
+
+}

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

@@ -0,0 +1,14 @@
+package com.ttstd.signaling.model;
+
+import lombok.Data;
+
+@Data
+public class ClientSdpMessage {
+    private String type;      // 类型: "offer", "answer", "ice", "reject", "accept"
+    private String sdp;
+
+    private String fromUser;
+    private String toUser;
+    private long timestamp;
+
+}

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

@@ -0,0 +1,12 @@
+package com.ttstd.signaling.model;
+
+import lombok.Data;
+
+@Data
+public class ServiceIceMessage {
+    private String sdpMid;
+    private int sdpMLineIndex;
+    private String sdp;
+    private String fromUser;
+    long timestamp;
+}

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

@@ -0,0 +1,11 @@
+package com.ttstd.signaling.model;
+
+import lombok.Data;
+
+@Data
+public class ServiceSdpMessage {
+    private String type;      // 类型: "offer", "answer", "ice", "reject", "accept"
+    private String sdp;
+    private String fromUser;
+    long timestamp;
+}

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

@@ -1,15 +0,0 @@
-package com.ttstd.signaling.model;
-
-import lombok.Data;
-import java.util.Map;
-
-@Data
-public class SignalingMessage {
-    private String type;          // 消息类型: offer, answer, candidate, join, leave
-    private String from;          // 发送者ID
-    private String to;            // 接收者ID
-    private String roomId;        // 房间ID
-    private Object sdp;           // SDP描述
-    private Object candidate;     // ICE候选
-    private Map<String, Object> payload; // 其他负载数据
-}

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,18 @@
 package com.ttstd.signaling.service;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.ttstd.signaling.config.HttpSessionConfigurator;
+import com.ttstd.signaling.bean.SignalingMessage;
-import com.ttstd.signaling.model.SignalMessage;
+import com.ttstd.signaling.model.WebRTCMessage;
-import com.ttstd.signaling.model.SignalingMessage;
-import jakarta.servlet.http.HttpSession;
-import jakarta.websocket.*;
-import jakarta.websocket.server.ServerEndpoint;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Component;
-
-import java.io.IOException;
-import java.util.concurrent.ConcurrentHashMap;
-
 import jakarta.websocket.*;
 import jakarta.websocket.server.PathParam;
 import jakarta.websocket.server.ServerEndpoint;
-import com.fasterxml.jackson.databind.ObjectMapper;
+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.io.IOException;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -33,29 +27,195 @@ public class WebRTCSignalingServer {
     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 RedisTemplate<String, Object> redisTemplate;
+
+    @Autowired
+    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) {
-        try {
+        logger.info("onMessage: 收到来自用户 {} 的消息: {}", userId, message);
-            // 解析收到的 JSON 消息
-            SignalMessage signal = objectMapper.readValue(message, SignalMessage.class);
-            String toUser = signal.getToUser();
-            Session targetSession = clients.get(toUser);
 
-            if (targetSession != null && targetSession.isOpen()) {
+        try {
-                // 转发信令（SDP, ICE, 拒绝/接受）
+            WebRTCMessage webRTCMessage = objectMapper.readValue(message, WebRTCMessage.class);
-                targetSession.getAsyncRemote().sendText(message);
+            if (webRTCMessage == null) {
-                logger.info("onMessage: 信令类型 [" + signal.getType() + "] 从 " + userId + " 发往 " + toUser);
+                logger.warn("onMessage: 无法解析消息，请检查消息格式。");
-            } else {
+                return;
-                sendErrorMessage(session, "目标用户不在线: " + toUser);
             }
+
+            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 {
+        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);
+        }
+
+        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);
+            }
+        } else {
+            logger.warn("onMessage: 收到消息但缺少 targetId 或 targetId 为空，无法转发或存储。消息内容: {}", webRTCMessage);
+            sendErrorMessage(session, userId, "消息格式错误，缺少目标用户ID。");
+        }
+    }
+
+    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);
+                }
+            } else {
+                sendErrorMessage(session, userId, "onMessage: 目标用户 {" + targetId + "} 不在线");
+                logger.info("onMessage: 目标用户 {} 不在线", targetId);
+            }
+        } else {
+            logger.warn("onMessage: 收到消息但缺少 targetId 或 targetId 为空，无法转发或存储。消息内容: {}", webRTCMessage);
+            sendErrorMessage(session, userId, "消息格式错误，缺少目标用户ID。");
         }
     }
 
@@ -67,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