优化登录
This commit is contained in:
10
DockerFile
Normal file
10
DockerFile
Normal file
@@ -0,0 +1,10 @@
|
||||
FROM eclipse-temurin:21-jdk-jammy
|
||||
MAINTAINER TongTongStudio <tongtongstudios@gmail.com>
|
||||
RUN mv /etc/apt/sources.list /etc/apt/sources.list.bak
|
||||
|
||||
# RUN mkdir -p /data/uploads/
|
||||
|
||||
VOLUME /tmp
|
||||
ADD target/*.jar app.jar
|
||||
EXPOSE 2310
|
||||
ENTRYPOINT ["java", "-jar", "/app.jar"]
|
||||
12
src/main/java/com/ttstd/signaling/model/SignalMessage.java
Normal file
12
src/main/java/com/ttstd/signaling/model/SignalMessage.java
Normal file
@@ -0,0 +1,12 @@
|
||||
package com.ttstd.signaling.model;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class SignalMessage {
|
||||
private String type; // 类型: "offer", "answer", "ice", "reject", "accept"
|
||||
private String fromUser; // 发送者ID
|
||||
private String toUser; // 接收者ID
|
||||
private Object data; // 数据载体 (SDP 描述或 ICE 候选信息)
|
||||
private String reason; // 用于拒绝时的原因说明
|
||||
}
|
||||
@@ -2,6 +2,7 @@ package com.ttstd.signaling.service;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.ttstd.signaling.config.HttpSessionConfigurator;
|
||||
import com.ttstd.signaling.model.SignalMessage;
|
||||
import com.ttstd.signaling.model.SignalingMessage;
|
||||
import jakarta.servlet.http.HttpSession;
|
||||
import jakarta.websocket.*;
|
||||
@@ -13,197 +14,63 @@ 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.springframework.stereotype.Component;
|
||||
import java.io.IOException;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
@Component
|
||||
@ServerEndpoint(value = "/signaling", configurator = HttpSessionConfigurator.class)
|
||||
@ServerEndpoint("/signaling/{userId}")
|
||||
public class WebRTCSignalingServer {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(WebRTCSignalingServer.class);
|
||||
Logger logger = LoggerFactory.getLogger(WebRTCSignalingServer.class);
|
||||
|
||||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
// 存储在线用户会话：key=userId, value=WebSocket会话
|
||||
private static ConcurrentHashMap<String, Session> onlineUsers = new ConcurrentHashMap<>();
|
||||
// 存储用户房间关系：key=userId, value=roomId
|
||||
private static ConcurrentHashMap<String, String> userRooms = new ConcurrentHashMap<>();
|
||||
// 存储房间用户：key=roomId, value=用户ID集合
|
||||
private static ConcurrentHashMap<String, ConcurrentHashMap<String, String>> roomUsers = new ConcurrentHashMap<>();
|
||||
|
||||
private Session session;
|
||||
private String userId;
|
||||
private HttpSession httpSession;
|
||||
// 用于存储在线用户的 Session
|
||||
private static final Map<String, Session> clients = new ConcurrentHashMap<>();
|
||||
private static final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
@OnOpen
|
||||
public void onOpen(Session session, EndpointConfig config) {
|
||||
this.session = session;
|
||||
this.httpSession = (HttpSession) config.getUserProperties()
|
||||
.get(HttpSession.class.getName());
|
||||
|
||||
// 从HTTP会话获取用户ID（实际项目中应从认证信息获取）
|
||||
this.userId = (String) this.httpSession.getAttribute("userId");
|
||||
if (this.userId == null) {
|
||||
this.userId = "user_" + session.getId().substring(0, 8);
|
||||
}
|
||||
|
||||
onlineUsers.put(this.userId, session);
|
||||
logger.info("用户 {} 连接成功, 当前在线用户数: {}", this.userId, onlineUsers.size());
|
||||
|
||||
// 发送连接成功消息
|
||||
sendMessage(createSignalingMessage("connected", this.userId, null, null, null, null));
|
||||
public void onOpen(Session session, @PathParam("userId") String userId) {
|
||||
clients.put(userId, session);
|
||||
System.out.println("用户连接: " + userId);
|
||||
}
|
||||
|
||||
@OnMessage
|
||||
public void onMessage(String message, Session session) {
|
||||
public void onMessage(String message, Session session, @PathParam("userId") String userId) {
|
||||
try {
|
||||
SignalingMessage signalingMessage = objectMapper.readValue(message, SignalingMessage.class);
|
||||
signalingMessage.setFrom(this.userId);
|
||||
processSignalingMessage(signalingMessage);
|
||||
// 解析收到的 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);
|
||||
System.out.println("信令类型 [" + signal.getType() + "] 从 " + userId + " 发往 " + toUser);
|
||||
} else {
|
||||
sendErrorMessage(session, "目标用户不在线: " + toUser);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
logger.error("消息处理错误: {}", e.getMessage());
|
||||
sendError("消息格式错误");
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
@OnClose
|
||||
public void onClose(Session session) {
|
||||
leaveRoom();
|
||||
onlineUsers.remove(this.userId);
|
||||
logger.info("用户 {} 断开连接, 剩余在线用户数: {}", this.userId, onlineUsers.size());
|
||||
public void onClose(@PathParam("userId") String userId) {
|
||||
clients.remove(userId);
|
||||
System.out.println("用户离开: " + userId);
|
||||
}
|
||||
|
||||
@OnError
|
||||
public void onError(Session session, Throwable error) {
|
||||
logger.error("WebSocket错误: {}", error.getMessage());
|
||||
onlineUsers.remove(this.userId);
|
||||
leaveRoom();
|
||||
error.printStackTrace();
|
||||
}
|
||||
|
||||
private void processSignalingMessage(SignalingMessage message) {
|
||||
switch (message.getType()) {
|
||||
case "join":
|
||||
handleJoinRoom(message);
|
||||
break;
|
||||
case "offer":
|
||||
case "answer":
|
||||
case "candidate":
|
||||
forwardMessageToTarget(message);
|
||||
break;
|
||||
case "leave":
|
||||
handleLeaveRoom(message);
|
||||
break;
|
||||
default:
|
||||
sendError("不支持的消息类型: " + message.getType());
|
||||
}
|
||||
}
|
||||
|
||||
private void handleJoinRoom(SignalingMessage message) {
|
||||
String roomId = message.getRoomId();
|
||||
if (roomId == null || roomId.trim().isEmpty()) {
|
||||
sendError("房间ID不能为空");
|
||||
return;
|
||||
}
|
||||
|
||||
// 离开之前的房间
|
||||
leaveRoom();
|
||||
|
||||
// 加入新房间
|
||||
userRooms.put(this.userId, roomId);
|
||||
roomUsers.computeIfAbsent(roomId, k -> new ConcurrentHashMap<>())
|
||||
.put(this.userId, this.userId);
|
||||
|
||||
logger.info("用户 {} 加入房间 {}", this.userId, roomId);
|
||||
|
||||
// 通知房间内其他用户
|
||||
broadcastToRoom(roomId, createSignalingMessage("user-joined", this.userId, null, roomId, null, null), this.userId);
|
||||
|
||||
// 发送加入成功消息
|
||||
sendMessage(createSignalingMessage("joined", this.userId, null, roomId, null, null));
|
||||
}
|
||||
|
||||
private void handleLeaveRoom(SignalingMessage message) {
|
||||
leaveRoom();
|
||||
sendMessage(createSignalingMessage("left", this.userId, null, null, null, null));
|
||||
}
|
||||
|
||||
private void leaveRoom() {
|
||||
String roomId = userRooms.get(this.userId);
|
||||
if (roomId != null) {
|
||||
userRooms.remove(this.userId);
|
||||
ConcurrentHashMap<String, String> users = roomUsers.get(roomId);
|
||||
if (users != null) {
|
||||
users.remove(this.userId);
|
||||
if (users.isEmpty()) {
|
||||
roomUsers.remove(roomId);
|
||||
} else {
|
||||
// 通知房间内其他用户
|
||||
broadcastToRoom(roomId, createSignalingMessage("user-left", this.userId, null, roomId, null, null), this.userId);
|
||||
}
|
||||
}
|
||||
logger.info("用户 {} 离开房间 {}", this.userId, roomId);
|
||||
}
|
||||
}
|
||||
|
||||
private void forwardMessageToTarget(SignalingMessage message) {
|
||||
String targetUserId = message.getTo();
|
||||
if (targetUserId == null) {
|
||||
sendError("目标用户ID不能为空");
|
||||
return;
|
||||
}
|
||||
|
||||
Session targetSession = onlineUsers.get(targetUserId);
|
||||
if (targetSession != null && targetSession.isOpen()) {
|
||||
try {
|
||||
targetSession.getBasicRemote().sendText(objectMapper.writeValueAsString(message));
|
||||
logger.debug("消息从 {} 转发到 {}", this.userId, targetUserId);
|
||||
} catch (IOException e) {
|
||||
logger.error("消息转发失败: {}", e.getMessage());
|
||||
onlineUsers.remove(targetUserId);
|
||||
}
|
||||
} else {
|
||||
sendError("目标用户不在线");
|
||||
}
|
||||
}
|
||||
|
||||
private void broadcastToRoom(String roomId, SignalingMessage message, String excludeUserId) {
|
||||
ConcurrentHashMap<String, String> users = roomUsers.get(roomId);
|
||||
if (users != null) {
|
||||
users.keySet().forEach(userId -> {
|
||||
if (!userId.equals(excludeUserId)) {
|
||||
Session userSession = onlineUsers.get(userId);
|
||||
if (userSession != null && userSession.isOpen()) {
|
||||
try {
|
||||
userSession.getBasicRemote().sendText(objectMapper.writeValueAsString(message));
|
||||
} catch (IOException e) {
|
||||
logger.error("广播消息失败: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private void sendMessage(SignalingMessage message) {
|
||||
try {
|
||||
this.session.getBasicRemote().sendText(objectMapper.writeValueAsString(message));
|
||||
} catch (IOException e) {
|
||||
logger.error("发送消息失败: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void sendError(String errorMessage) {
|
||||
SignalingMessage errorMsg = createSignalingMessage("error", "system", this.userId, null, null, null);
|
||||
errorMsg.setPayload(java.util.Map.of("message", errorMessage));
|
||||
sendMessage(errorMsg);
|
||||
}
|
||||
|
||||
private SignalingMessage createSignalingMessage(String type, String from, String to,
|
||||
String roomId, Object sdp, Object candidate) {
|
||||
SignalingMessage message = new SignalingMessage();
|
||||
message.setType(type);
|
||||
message.setFrom(from);
|
||||
message.setTo(to);
|
||||
message.setRoomId(roomId);
|
||||
message.setSdp(sdp);
|
||||
message.setCandidate(candidate);
|
||||
return message;
|
||||
private void sendErrorMessage(Session session, String errorMsg) {
|
||||
session.getAsyncRemote().sendText("{\"type\":\"error\", \"message\":\"" + errorMsg + "\"}");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user