【message】消息模块增加websocket

kernel-s-message/message-sdk-websocket/README.md

@@ -0,0 +1 @@
+系统消息websocket的sdk，用于将消息发送给在线用户，并提供相应接口

kernel-s-message/message-sdk-websocket/pom.xml

@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>cn.stylefeng.roses</groupId>
+        <artifactId>kernel-s-message</artifactId>
+        <version>1.0.0</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>message-sdk-websocket</artifactId>
+
+    <packaging>jar</packaging>
+
+    <dependencies>
+
+        <!--消息模块的api-->
+        <dependency>
+            <groupId>cn.stylefeng.roses</groupId>
+            <artifactId>message-api</artifactId>
+            <version>1.0.0</version>
+        </dependency>
+
+        <!--websocket 依赖-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
+
+    </dependencies>
+
+</project>

kernel-s-message/message-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/message/websocket/manager/WebSocketManager.java

@@ -0,0 +1,95 @@
+package cn.stylefeng.roses.kernel.message.websocket.manager;
+
+import org.springframework.util.CollectionUtils;
+
+import javax.websocket.Session;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author liuhq
+ */
+public class WebSocketManager {
+    private static final ConcurrentHashMap<Long, List<Session>> userIdSessionMap = new ConcurrentHashMap<>();
+
+    /**
+     * 添加用户ID相关的Session
+     *
+     * @param userId  用户id
+     * @param session 用户websocketSession
+     * @author liuhanqing
+     * @date 2021/1/24 22:08
+     */
+    public static void add(Long userId, Session session) {
+        userIdSessionMap.computeIfAbsent(userId, v -> new ArrayList<>()).add(session);
+    }
+
+    /**
+     * 根据用户ID获取Session
+     *
+     * @param userId 用户id
+     * @return List<Session>  用户websocketSession集合
+     * @author liuhanqing
+     * @date 2021/1/24 22:10
+     */
+    public static List<Session> getSessionByUserId(Long userId) {
+        return userIdSessionMap.get(userId);
+    }
+
+    /**
+     * 移除失效的Session
+     *
+     * @param userId   用户id
+     * @param session 用户websocketSession
+     * @author liuhanqing
+     * @date 2021/1/24 22:11
+     */
+    public static void removeSession(Long userId, Session session) {
+        if (session == null) {
+            return;
+        }
+        List<Session> webSessoin = userIdSessionMap.get(userId);
+        if (webSessoin == null || CollectionUtils.isEmpty(webSessoin)) {
+            return;
+        }
+        webSessoin.remove(session);
+    }
+
+    /**
+     * 获取链接用户集合
+     *
+     * @author liuhanqing
+     * @date 2021/1/24 22:11
+     */
+    public static Set<Long> getUserList() {
+        return userIdSessionMap.keySet();
+    }
+
+    /**
+     * 发送消息
+     *
+     * @param userId  用户id
+     * @param message 消息
+     * @author liuhanqing
+     * @date 2021/1/24 22:11
+     */
+    public static void sendMessage(Long userId, String message){
+        for(Session userSession: getSessionByUserId(userId)){
+            userSession.getAsyncRemote().sendText(message);
+        }
+    }
+    /**
+     * 发送消息
+     *
+     * @param message 消息
+     * @author liuhanqing
+     * @date 2021/1/24 22:11
+     */
+    public static void sendMessageToAll(String message){
+        for (Long userId : WebSocketManager.getUserList()) {
+            sendMessage(userId, message);
+        }
+    }
+}

kernel-s-message/message-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/message/websocket/server/WebSocketEndpoint.java

@@ -0,0 +1,78 @@
+package cn.stylefeng.roses.kernel.message.websocket.server;
+
+import cn.stylefeng.roses.kernel.message.websocket.manager.WebSocketManager;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.websocket.*;
+import javax.websocket.server.PathParam;
+import javax.websocket.server.ServerEndpoint;
+
+/**
+ * websocket服务端
+ *
+ * @author liuhanqing
+ * @date 2021/1/24 22:26
+ */
+@Slf4j
+@Component
+@ServerEndpoint("/message/websocket/{userId}")
+public class WebSocketEndpoint {
+
+
+    /**
+     * 连接建立成功后调用
+     *
+     * @param userId  用户id
+     * @param session 用户websocketSession
+     * @author liuhanqing
+     * @date 2021/1/24 22:27
+     */
+    @OnOpen
+    public void onOpen(@PathParam(value = "userId") Long userId, Session session) {
+        // 添加到链接管理
+        WebSocketManager.add(userId, session);
+        // 返回消息
+        session.getAsyncRemote().sendText("WebSocket连接成功");
+    }
+
+    /**
+     * 连接关闭时调用
+     *
+     * @author liuhanqing
+     * @date 2021/1/24 22:29
+     */
+    @OnClose
+    public void onClose(@PathParam(value = "userId") Long userId, Session session) {
+        // 从map中删除
+        WebSocketManager.removeSession(userId, session);
+    }
+
+    /**
+     * 收到客户端消息后调用
+     *
+     * @param message 客户端发送过来的消息
+     * @param session 用户websocketSession
+     * @author liuhanqing
+     * @date 2021/1/24 22:29
+     */
+    @OnMessage
+    public void onMessage(String message, Session session) {
+        log.info("来自客户端的消息:" + message);
+    }
+
+    /**
+     * 发生错误时回调
+     *
+     * @param session 用户信息
+     * @param error   错误
+     * @author liuhanqing
+     * @date 2021/1/24 22:29
+     */
+    @OnError
+    public void onError(Session session, Throwable error) {
+        log.error("WebSocket发生错误");
+        error.printStackTrace();
+    }
+
+}

kernel-s-message/message-spring-boot-starter/pom.xml

@@ -32,6 +32,11 @@
             <version>1.0.0</version>
         </dependency>
 
+        <dependency>
+            <groupId>cn.stylefeng.roses</groupId>
+            <artifactId>message-sdk-websocket</artifactId>
+            <version>1.0.0</version>
+        </dependency>
 
     </dependencies>
 

kernel-s-message/message-spring-boot-starter/src/main/java/cn/stylefeng/roses/kernel/message/starter/GunsMessageAutoConfiguration.java

@@ -1,6 +1,9 @@
 package cn.stylefeng.roses.kernel.message.starter;
 
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
 
 /**
  * 系统消息的自动配置
@@ -11,6 +14,17 @@ import org.springframework.context.annotation.Configuration;
 @Configuration
 public class GunsMessageAutoConfiguration {
 
-
+    /**
+     *  开启WebSocket功能
+     *
+     * @return serverEndpointExporter
+     * @author liuhanqing
+     * @date 2021/01/24 22:09
+     */
+    @Bean
+    @ConditionalOnMissingBean(ServerEndpointExporter.class)
+    public ServerEndpointExporter serverEndpointExporter() {
+        return new ServerEndpointExporter();
+    }
 
 }

kernel-s-message/pom.xml

@@ -19,6 +19,7 @@
         <module>message-api</module>
         <module>message-business</module>
         <module>message-sdk-db</module>
+        <module>message-sdk-websocket</module>
         <module>message-spring-boot-starter</module>
     </modules>