message-service 分别使用 EventReq 和 EventResp 来表示 client 向 server 发送的事件和 server 向 client 返回的事件

message/message-service/src/main/java/cn/reghao/tnb/message/app/rabbit/RabbitListeners.java

@@ -40,7 +40,7 @@ public class RabbitListeners {
         if (wsConnection.getUserSession(receiverId) != null) {
             log.info("send {} -> {}", receiverId, content);
             try {
-                wsConnection.sendText(receiverId+"", msg);
+                wsConnection.sendChatMessage(receiverId+"", chatPayload);
             } catch (IOException e) {
                 e.printStackTrace();
             }

message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/WsConnection.java

@@ -11,7 +11,8 @@ import cn.reghao.tnb.message.api.model.EventType;
 import cn.reghao.tnb.message.api.model.resp.EventMessageResp;
 import cn.reghao.tnb.message.app.rabbit.RabbitProducer;
 import cn.reghao.tnb.message.app.ws.msg.ChatPayload;
-import cn.reghao.tnb.message.app.ws.msg.EventMessage;
+import cn.reghao.tnb.message.app.ws.msg.EventReq;
+import cn.reghao.tnb.message.app.ws.msg.EventResp;
 import cn.reghao.tnb.message.app.ws.msg.PingPayload;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.dubbo.config.annotation.DubboReference;
@@ -79,12 +80,12 @@ public class WsConnection {
     }
 
     public void sendPingPayload(WebSocketSession session) throws IOException {
-        PingPayload pingPayload = new PingPayload(60, 300);
-        EventMessage eventMessage = new EventMessage();
-        eventMessage.setEvent("connect");
-        eventMessage.setPayload(pingPayload);
+        PingPayload pingPayload = new PingPayload(60, 120);
+        EventResp eventResp = new EventResp();
+        eventResp.setEvent("connect");
+        eventResp.setPayload(pingPayload);
 
-        TextMessage textMessage = new TextMessage(JsonConverter.objectToJson(pingPayload));
+        TextMessage textMessage = new TextMessage(JsonConverter.objectToJson(eventResp));
         session.sendMessage(textMessage);
     }
 
@@ -155,11 +156,11 @@ public class WsConnection {
             return;
         }
 
-        EventMessage eventMessage = new EventMessage();
-        eventMessage.setEvent("event_talk");
-        eventMessage.setPayload(chatPayload);
+        EventResp eventResp = new EventResp();
+        eventResp.setEvent("event_talk");
+        eventResp.setPayload(chatPayload);
 
-        TextMessage textMessage = new TextMessage(JsonConverter.objectToJson(eventMessage));
+        TextMessage textMessage = new TextMessage(JsonConverter.objectToJson(eventResp));
         session.sendMessage(textMessage);
     }
 

message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/handler/ChatHandler.java

@@ -4,8 +4,8 @@ import cn.reghao.jutil.jdk.serializer.JdkSerializer;
 import cn.reghao.jutil.jdk.serializer.JsonConverter;
 import cn.reghao.tnb.message.api.model.EventType;
 import cn.reghao.tnb.message.app.ws.WsConnection;
+import cn.reghao.tnb.message.app.ws.msg.EventReq;
 import cn.reghao.tnb.message.app.ws.service.ChatService;
-import cn.reghao.tnb.message.app.ws.msg.EventPayload;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.*;
@@ -81,8 +81,8 @@ public class ChatHandler implements WebSocketHandler {
         String payload = textMessage.getPayload();
         try {
             // TODO gson 会将 Long 反序列化为 Double
-            EventPayload eventPayload = JsonConverter.jsonToObject(payload, EventPayload.class);
-            String event = eventPayload.getEvent();
+            EventReq eventReq = JsonConverter.jsonToObject(payload, EventReq.class);
+            String event = eventReq.getEvent();
             switch (EventType.valueOf(event)) {
                 case heartbeat:
                     wsConnection.sendHeartbeatPong(webSocketSession);
@@ -94,7 +94,7 @@ public class ChatHandler implements WebSocketHandler {
                     log.info("talk_keyboard event");
                     break;
                 case event_talk:
-                    chatService.onEventTalk(payload);
+                    chatService.onEventTalk(eventReq.getData());
                     break;
                 case event_talk_revoke:
                     log.info("talk_revoke event");

message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/handler/ProgressHandler.java

@@ -4,7 +4,7 @@ import cn.reghao.jutil.jdk.serializer.JsonConverter;
 import cn.reghao.tnb.message.api.dto.MediaProgress;
 import cn.reghao.tnb.message.api.model.EventType;
 import cn.reghao.tnb.message.app.ws.WsConnection;
-import cn.reghao.tnb.message.app.ws.msg.EventPayload;
+import cn.reghao.tnb.message.app.ws.msg.EventReq;
 import cn.reghao.tnb.message.app.ws.service.ProgressService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -77,15 +77,15 @@ public class ProgressHandler implements WebSocketHandler {
         String sessionId = webSocketSession.getId();
         String payload = textMessage.getPayload();
         try {
-            EventPayload eventPayload = JsonConverter.jsonToObject(payload, EventPayload.class);
-            String event = eventPayload.getEvent();
+            EventReq eventReq = JsonConverter.jsonToObject(payload, EventReq.class);
+            String event = eventReq.getEvent();
             switch (EventType.valueOf(event)) {
                 case heartbeat:
                     wsConnection.sendHeartbeatPong(webSocketSession);
                     break;
                 case media_progress:
-                    String jsonPayload = eventPayload.getPayload();
-                    MediaProgress mediaProgress = JsonConverter.jsonToObject(jsonPayload, MediaProgress.class);
+                    String jsonData = eventReq.getData();
+                    MediaProgress mediaProgress = JsonConverter.jsonToObject(jsonData, MediaProgress.class);
                     progressService.putMediaProgress(sessionId, mediaProgress);
                     break;
                 default:

message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/msg/ChatPayload.java

@@ -2,12 +2,15 @@ package cn.reghao.tnb.message.app.ws.msg;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
 
 /**
  * @author reghao
  * @date 2024-11-29 10:39:54
  */
-@AllArgsConstructor
+@NoArgsConstructor
+@Setter
 @Getter
 public class ChatPayload {
     private Long senderId;

message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/msg/EventPayload.java → message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/msg/EventReq.java

@@ -5,12 +5,12 @@ import lombok.Setter;
 
 /**
  * @author reghao
- * @date 2024-12-02 11:07:53
+ * @date 2024-11-28 17:57:37
  */
 @Setter
 @Getter
-public class EventPayload {
+public class EventReq {
     private String event;
     // JSON 字符串
-    private String payload;
+    private String data;
 }

message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/msg/EventMessage.java → message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/msg/EventResp.java

@@ -5,11 +5,11 @@ import lombok.Setter;
 
 /**
  * @author reghao
- * @date 2024-11-28 17:57:37
+ * @date 2024-12-02 16:15:33
  */
 @Setter
 @Getter
-public class EventMessage {
+public class EventResp {
     private String event;
     private Object payload;
 }

message/message-service/src/main/java/cn/reghao/tnb/message/app/ws/service/ChatService.java

@@ -1,20 +1,13 @@
 package cn.reghao.tnb.message.app.ws.service;
 
 import cn.reghao.jutil.jdk.serializer.JsonConverter;
-import cn.reghao.tnb.account.api.constant.TokenType;
-import cn.reghao.tnb.account.api.dto.AuthedAccount;
 import cn.reghao.tnb.account.api.iface.AccountQuery;
 import cn.reghao.tnb.message.api.model.EventType;
 import cn.reghao.tnb.message.api.model.resp.EventMessageResp;
 import cn.reghao.tnb.message.api.model.resp.EvtLoginResp;
-import cn.reghao.tnb.message.app.rabbit.RabbitProducer;
 import cn.reghao.tnb.message.app.ws.WsConnection;
 import cn.reghao.tnb.message.app.ws.msg.ChatPayload;
-import cn.reghao.tnb.message.app.ws.msg.EventMessage;
 import cn.reghao.tnb.user.api.iface.UserContactService;
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import com.google.gson.JsonObject;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.springframework.stereotype.Component;
@@ -22,9 +15,7 @@ import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
 
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 /**
  * @author reghao
@@ -38,29 +29,12 @@ public class ChatService {
     @DubboReference(check = false, retries = 0)
     private UserContactService userContactService;
 
-    private final Map<String, Long> sessionMap = new HashMap<>();
-    private final Map<Long, WebSocketSession> userMap = new HashMap<>();
-    private final Gson gson = new GsonBuilder().create();
     private final WsConnection wsConnection;
-    private RabbitProducer rabbitProducer;
 
     public ChatService(WsConnection wsConnection) {
         this.wsConnection = wsConnection;
     }
 
-    public void online(WebSocketSession session) throws IOException {
-        String query = session.getUri().getQuery();
-        String jwtToken = query.replace("token=", "");
-        AuthedAccount authedAccount = accountQuery.getAuthedAccount(TokenType.token.getValue(), jwtToken);
-        if (authedAccount != null) {
-            long userId = authedAccount.getUserId();
-            onLoginEvent(userId, true);
-            sessionMap.put(session.getId(), userId);
-            userMap.put(userId, session);
-            log.info("{} 在线", userId);
-        }
-    }
-
     public void onLoginEvent(long userId, boolean online) {
         EvtLoginResp evtLoginResp = new EvtLoginResp(userId, online);
         EventMessageResp<EvtLoginResp> resp = new EventMessageResp<>(EventType.event_login, evtLoginResp);
@@ -69,7 +43,7 @@ public class ChatService {
         userContactService.setOnline(userId, online);
         List<Long> friendIds = userContactService.getOnlineFriends(userId);
         for (Long friendId : friendIds) {
-            WebSocketSession session = userMap.get(friendId);
+            WebSocketSession session = wsConnection.getUserSession(friendId);
             if (session != null) {
                 try {
                     session.sendMessage(textMessage);
@@ -88,6 +62,6 @@ public class ChatService {
             return;
         }
 
-        wsConnection.sendText(receiverId+"", payload);
+        wsConnection.sendChatMessage(receiverId+"", chatPayload);
     }
 }