package com.xiaopeng.pushMessage.webSocket.handler;
|
|
import com.alibaba.fastjson.JSON;
|
import com.alibaba.fastjson.JSONObject;
|
import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
|
import io.netty.channel.ChannelHandlerContext;
|
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Component;
|
import org.springframework.util.StringUtils;
|
import org.springframework.web.socket.CloseStatus;
|
import org.springframework.web.socket.TextMessage;
|
import org.springframework.web.socket.WebSocketSession;
|
import org.springframework.web.socket.handler.TextWebSocketHandler;
|
|
import java.time.LocalDateTime;
|
import java.util.UUID;
|
|
/**
|
* Spring WebSocket 消息处理器
|
* 路径: /websocket，共用 Spring Boot HTTP 端口
|
* 替代原 Netty 独立端口方案
|
*/
|
@Component
|
@Slf4j
|
public class PushMessageWebSocketHandler extends TextWebSocketHandler {
|
|
@Autowired
|
private WebSocketUserConnectionManager userConnectionManager;
|
|
@Autowired
|
private WebSocketAuthHandler authHandler;
|
|
// ==================== 连接生命周期 ====================
|
|
@Override
|
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
|
log.info("WebSocket 连接建立，sessionId: {}", session.getId());
|
}
|
|
@Override
|
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
|
log.info("WebSocket 连接关闭，sessionId: {}, status: {}", session.getId(), status);
|
userConnectionManager.removeUserConnection(session);
|
}
|
|
@Override
|
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
|
log.error("WebSocket 传输异常，sessionId: {}, error: {}", session.getId(), exception.getMessage(), exception);
|
if (session.isOpen()) {
|
session.close(CloseStatus.SERVER_ERROR);
|
}
|
}
|
|
// ==================== 消息处理 ====================
|
|
@Override
|
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
|
String payload = message.getPayload();
|
log.info("[服务器] 收到消息 -> {}", payload);
|
|
try {
|
JSONObject messageObj = JSON.parseObject(payload);
|
String messageType = messageObj.getString("type");
|
|
if ("auth".equals(messageType)) {
|
handleUserAuth(session, messageObj);
|
} else if ("ping".equals(messageType)) {
|
handlePing(session);
|
} else {
|
handleBusinessMessage(session, messageObj);
|
}
|
} catch (Exception e) {
|
log.debug("消息非 JSON 格式，按普通文本处理: {}", payload);
|
sendText(session, "服务器时间: " + LocalDateTime.now() + "->" + payload);
|
}
|
}
|
|
// ==================== 业务处理 ====================
|
|
/**
|
* 处理用户认证消息
|
*/
|
private void handleUserAuth(WebSocketSession session, JSONObject messageObj) {
|
try {
|
WebSocketAuthHandler.UserAuthInfo authInfo = authHandler.extractAuthInfo(messageObj);
|
String appUserId = authInfo.getAppUserId();
|
String token = authInfo.getToken();
|
|
if (!StringUtils.hasText(appUserId)) {
|
sendAuthFailure(session, "用户ID不能为空");
|
return;
|
}
|
|
WebSocketAuthHandler.AuthResult authResult = authHandler.authenticateUser(appUserId, token);
|
if (authResult.isSuccess()) {
|
userConnectionManager.addUserConnection(appUserId, session);
|
|
// TODO
|
String uuid = UUID.randomUUID().toString();
|
JSONObject response = new JSONObject();
|
response.put("type", "auth_success");
|
response.put("appUserId", appUserId);
|
response.put("nickName", uuid);
|
response.put("message", "认证成功");
|
response.put("timestamp", LocalDateTime.now().toString());
|
|
sendText(session, response.toJSONString());
|
log.info("用户 {} ({}) 认证成功，sessionId: {}", appUserId, uuid, session.getId());
|
} else {
|
sendAuthFailure(session, authResult.getMessage());
|
log.warn("用户 {} 认证失败: {}, sessionId: {}", appUserId, authResult.getMessage(), session.getId());
|
}
|
} catch (Exception e) {
|
log.error("处理用户认证时发生错误", e);
|
sendAuthFailure(session, "认证过程中发生错误: " + e.getMessage());
|
}
|
}
|
|
private void sendAuthFailure(WebSocketSession session, String message) {
|
JSONObject response = new JSONObject();
|
response.put("type", "auth_failed");
|
response.put("message", message);
|
response.put("timestamp", LocalDateTime.now().toString());
|
sendText(session, response.toJSONString());
|
}
|
|
/**
|
* 处理心跳消息
|
*/
|
private void handlePing(WebSocketSession session) {
|
String appUserId = userConnectionManager.getUserIdBySessionId(session.getId());
|
|
JSONObject response = new JSONObject();
|
response.put("type", "pong");
|
response.put("appUserId", appUserId);
|
response.put("timestamp", LocalDateTime.now().toString());
|
log.info("心跳用户id：{}", appUserId);
|
|
sendText(session, response.toJSONString());
|
}
|
|
/**
|
* 处理业务消息
|
*/
|
private void handleBusinessMessage(WebSocketSession session, JSONObject messageObj) {
|
String appUserId = userConnectionManager.getUserIdBySessionId(session.getId());
|
|
if (appUserId != null) {
|
log.info("收到用户 {} 的业务消息: {}", appUserId, messageObj.toJSONString());
|
|
JSONObject response = new JSONObject();
|
response.put("type", "message_response");
|
response.put("appUserId", appUserId);
|
response.put("originalMessage", messageObj);
|
response.put("serverTime", LocalDateTime.now().toString());
|
|
sendText(session, response.toJSONString());
|
} else {
|
log.warn("收到未认证连接的业务消息: {}", messageObj.toJSONString());
|
|
JSONObject response = new JSONObject();
|
response.put("type", "auth_required");
|
response.put("message", "请先进行用户认证");
|
response.put("timestamp", LocalDateTime.now().toString());
|
|
sendText(session, response.toJSONString());
|
}
|
}
|
|
// ==================== 工具方法 ====================
|
|
/**
|
* 发送文本消息（线程安全）
|
*/
|
private void sendText(WebSocketSession session, String text) {
|
try {
|
if (session.isOpen()) {
|
synchronized (session) {
|
session.sendMessage(new TextMessage(text));
|
}
|
}
|
} catch (Exception e) {
|
log.error("发送消息失败，sessionId: {}, error: {}", session.getId(), e.getMessage());
|
}
|
}
|
}
|
