feat:添加个人用户推送

Author: pedro

src/main/java/io/github/talelin/merak/modules/message/MessageConsts.java

@@ -0,0 +1,8 @@
+package io.github.talelin.merak.modules.message;
+
+/**
+ * websocket 模块常量
+ */
+public class MessageConsts {
+    public static final String USER_KEY = "user";
+}

src/main/java/io/github/talelin/merak/modules/message/README.md

@@ -188,7 +188,7 @@ WsHandler 是 lin-cms 提供给开发者的消息推送接口，该接口有许
 
 ### sendMessage
 
-给某个会话发送消息，有两个重载方法，如下：
+给某个会话或某个用户发送消息，有四个重载方法，如下：
 
 ```
 /**
@@ -208,10 +208,31 @@ void sendMessage(WebSocketSession session, String message) throws IOException;
  * @throws IOException 发送io异常
  */
 void sendMessage(WebSocketSession session, TextMessage message) throws IOException;
+
+/**
+ * 发送消息
+ *
+ * @param userId  用户id
+ * @param message 要发送的消息
+ * @throws IOException 发送io异常
+ */
+void sendMessage(Long userId, TextMessage message) throws IOException;
+
+/**
+ * 发送消息
+ *
+ * @param userId  用户id
+ * @param message 要发送的消息
+ * @throws IOException 发送io异常
+ */
+void sendMessage(Long userId, String message) throws IOException;
 ```
 
 方法的第二个参数可以直接接收字符串，也可以接收 TextMessage 类型的消息。
 
+当给用户发送消息时，只需要传入用户 id 即可，但一定要确保该用户通过令牌进行了
+连接。
+
 ### broadCast
 
 广播，给所有连接用户发送消息，也有两个重载方法，第一个直接发送字符串，

src/main/java/io/github/talelin/merak/modules/message/WebSocketInterceptor.java

@@ -18,6 +18,8 @@
 import java.nio.charset.Charset;
 import java.util.Map;
 
+import static io.github.talelin.merak.modules.message.MessageConsts.USER_KEY;
+
 @SuppressWarnings("Duplicates")
 public class WebSocketInterceptor implements HandshakeInterceptor {
     @Autowired
@@ -58,7 +60,7 @@ public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse res
                 writeMessageToBody(response, "user is not found");
                 return false;
             }
-            attributes.put("user", user);
+            attributes.put(USER_KEY, user);
             return true;
         }
         return false;

src/main/java/io/github/talelin/merak/modules/message/WsHandler.java

@@ -39,6 +39,24 @@ public interface WsHandler {
      */
     void sendMessage(WebSocketSession session, String message) throws IOException;
 
+    /**
+     * 发送消息
+     *
+     * @param userId  用户id
+     * @param message 要发送的消息
+     * @throws IOException 发送io异常
+     */
+    void sendMessage(Long userId, TextMessage message) throws IOException;
+
+    /**
+     * 发送消息
+     *
+     * @param userId  用户id
+     * @param message 要发送的消息
+     * @throws IOException 发送io异常
+     */
+    void sendMessage(Long userId, String message) throws IOException;
+
     /**
      * 发送消息
      *

src/main/java/io/github/talelin/merak/modules/message/WsHandlerImpl.java

@@ -7,9 +7,12 @@
 import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
 
+import static io.github.talelin.merak.modules.message.MessageConsts.USER_KEY;
+
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -47,6 +50,29 @@ public void sendMessage(WebSocketSession session, String message) throws IOExcep
         this.sendMessage(session, new TextMessage(message));
     }
 
+    @Override
+    public void sendMessage(Long userId, TextMessage message) throws IOException {
+        Optional<WebSocketSession> userSession = sessions.stream().filter(session -> {
+            if (!session.isOpen()) {
+                return false;
+            }
+            Map<String, Object> attributes = session.getAttributes();
+            if (!attributes.containsKey(USER_KEY)) {
+                return false;
+            }
+            UserDO user = (UserDO) attributes.get(USER_KEY);
+            return user.getId().equals(userId);
+        }).findFirst();
+        if (userSession.isPresent()) {
+            userSession.get().sendMessage(message);
+        }
+    }
+
+    @Override
+    public void sendMessage(Long userId, String message) throws IOException {
+        sendMessage(userId, new TextMessage(message));
+    }
+
     @Override
     public void sendMessage(WebSocketSession session, TextMessage message) throws IOException {
         session.sendMessage(message);
@@ -82,9 +108,9 @@ public void broadCastToGroup(Long groupId, TextMessage message) throws IOExcepti
             if (!session.isOpen())
                 continue;
             Map<String, Object> attributes = session.getAttributes();
-            if (!attributes.containsKey("user"))
+            if (!attributes.containsKey(USER_KEY))
                 continue;
-            UserDO user = (UserDO) attributes.get("user");
+            UserDO user = (UserDO) attributes.get(USER_KEY);
             boolean matched = userIds.stream().anyMatch(id -> id.equals(user.getId()));
             if (!matched)
                 continue;