Merge pull request #2105 from dataease/pr@dev@feat_websocket_msg

backend/pom.xml

@@ -36,12 +36,12 @@
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
-            <exclusions>
+            <!--<exclusions>
                 <exclusion>
                     <artifactId>spring-boot-starter-tomcat</artifactId>
                     <groupId>org.springframework.boot</groupId>
                 </exclusion>
-            </exclusions>
+            </exclusions>-->
         </dependency>
 
         <dependency>

backend/src/main/java/io/dataease/auth/service/impl/ShiroServiceImpl.java

@@ -42,6 +42,7 @@ public class ShiroServiceImpl implements ShiroService {
         filterChainDefinitionMap.put("/index.html", ANON);
         filterChainDefinitionMap.put("/link.html", ANON);
         filterChainDefinitionMap.put("/board/**", ANON);
+        filterChainDefinitionMap.put("/websocket/**", "anon");
 
         // 获取主题信息
         filterChainDefinitionMap.put("/plugin/theme/themes", ANON);

backend/src/main/java/io/dataease/websocket/ServerEndpointConfigurator.java

-package io.dataease.websocket;
-
-
-import javax.websocket.HandshakeResponse;
-import javax.websocket.server.HandshakeRequest;
-import javax.websocket.server.ServerEndpointConfig;
-
-public class ServerEndpointConfigurator extends ServerEndpointConfig.Configurator {
-    @Override
-    public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) {
-        super.modifyHandshake(sec, request, response);
-    }
-}

backend/src/main/java/io/dataease/websocket/WebSocketServer.java

-package io.dataease.websocket;
-
-import org.springframework.stereotype.Component;
-
-import javax.websocket.*;
-import javax.websocket.server.ServerEndpoint;
-import java.io.IOException;
-
-@ServerEndpoint(value = "/socket", configurator = ServerEndpointConfigurator.class)
-@Component
-public class WebSocketServer {
-    @OnOpen
-    public void onOpen(Session session) throws IOException {
-
-    }
-
-    @OnMessage
-    public void onMessage(Session session, String message) throws IOException {
-
-    }
-
-    @OnClose
-    public void onClose(Session session) throws IOException {
-
-    }
-
-    @OnError
-    public void onError(Session session, Throwable throwable) {
-        throwable.printStackTrace();
-    }
-}

backend/src/main/java/io/dataease/websocket/aop/WSTrigger.java

+package io.dataease.websocket.aop;
+
+import io.dataease.websocket.entity.WsMessage;
+import io.dataease.websocket.service.WsService;
+import org.apache.commons.lang3.ArrayUtils;
+import org.aspectj.lang.JoinPoint;
+import org.aspectj.lang.annotation.AfterReturning;
+import org.aspectj.lang.annotation.Aspect;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.Optional;
+
+@Aspect
+@Component
+public class WSTrigger {
+
+    @Autowired
+    private WsService wsService;
+
+    @AfterReturning(value = "execution(* io.dataease.service.message.service.strategy.SendStation.sendMsg(..))")
+    public void after(JoinPoint point) {
+        Object[] args = point.getArgs();
+        Optional.ofNullable(args).ifPresent(objs -> {
+            if (ArrayUtils.isEmpty(objs)) return;
+            Object arg = args[0];
+            Long userId = (Long) arg;
+            WsMessage message = new WsMessage(userId, "/web-msg-topic", "refresh");
+            wsService.releaseMessage(message);
+        });
+
+    }
+}

backend/src/main/java/io/dataease/websocket/config/WsConfig.java

+package io.dataease.websocket.config;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.simp.config.MessageBrokerRegistry;
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
+import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration;
+
+@Configuration
+@EnableWebSocketMessageBroker
+public class WsConfig implements WebSocketMessageBrokerConfigurer {
+
+
+
+    @Override
+    public void registerStompEndpoints(StompEndpointRegistry registry) {
+        registry.addEndpoint("/websocket").setAllowedOriginPatterns("*").withSockJS();
+    }
+
+    @Override
+    public void configureMessageBroker(MessageBrokerRegistry registry) {
+        registry.enableSimpleBroker("/topic", "/user");
+        registry.setUserDestinationPrefix("/user");
+    }
+
+    @Override
+    public void configureWebSocketTransport(WebSocketTransportRegistration registry) {
+        registry.setMessageSizeLimit(8192) //设置消息字节数大小
+                .setSendBufferSizeLimit(8192)//设置消息缓存大小
+                .setSendTimeLimit(10000); //设置消息发送时间限制毫秒
+    }
+}

backend/src/main/java/io/dataease/websocket/entity/WsMessage.java

+package io.dataease.websocket.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class WsMessage<T> implements Serializable {
+    private Long userId;
+
+    private String topic;
+
+    private T data;
+
+
+}
+

backend/src/main/java/io/dataease/websocket/service/WsService.java

+package io.dataease.websocket.service;
+
+import io.dataease.websocket.entity.WsMessage;
+import org.apache.commons.lang3.ObjectUtils;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.Optional;
+
+@Component
+public class WsService {
+
+
+
+    @Resource
+    private SimpMessagingTemplate messagingTemplate;
+
+
+
+
+    public void releaseMessage (List<WsMessage> wsMessages) {
+        Optional.ofNullable(wsMessages).ifPresent(messages -> {
+            messages.forEach(this::releaseMessage);
+        });
+    }
+
+    public void releaseMessage(WsMessage wsMessage){
+        if(ObjectUtils.isEmpty(wsMessage) || ObjectUtils.isEmpty(wsMessage.getUserId()) ||  ObjectUtils.isEmpty(wsMessage.getTopic())) return;
+
+
+        messagingTemplate.convertAndSendToUser(String.valueOf(wsMessage.getUserId()), wsMessage.getTopic(),wsMessage.getData());
+
+
+    }
+
+
+}

frontend/package.json

@@ -45,6 +45,8 @@
     "normalize.css": "7.0.0",
     "nprogress": "0.2.0",
     "screenfull": "4.2.0",
+    "sockjs-client": "^1.6.0",
+    "stompjs": "^2.3.3",
     "svg-sprite-loader": "4.1.3",
     "svgo": "1.2.2",
     "tinymce": "^5.8.2",

frontend/src/components/Notification/index.vue

 <template>
 
   <el-popover
-
     v-model="visible"
     width="350"
     trigger="click"
@@ -9,7 +8,7 @@
     style="display: flex;align-items: center;"
     class="international"
   >
-    <div>
+    <div v-loading="loading">
       <div style="height: 30px;">
         <div style="float: left;font-size:16px;font-weight:bold;">
           <span>{{ $t('webmsg.web_msg') }}</span>
@@ -49,17 +48,10 @@
       </div>
     </div>
     <div slot="reference">
-      <el-badge :value="count || paginationConfig.total" :hidden="!count && !paginationConfig.total" :max="99" class="item">
+      <el-badge :value="visible ? paginationConfig.total : count" :hidden="!count && !paginationConfig.total" :max="99" class="item">
         <svg-icon class-name="notification" icon-class="notification" />
       </el-badge>
-      <!-- <div>
-        <svg-icon
-          class-name="notification"
-          icon-class="notification"
-        />
-        <span v-if="count || paginationConfig.total" class="msg-number">{{ count || paginationConfig.total }}</span>
-      </div>
-    </div> -->
+
     </div></el-popover>
 </template>
 
@@ -81,7 +73,8 @@ export default {
         total: 0
       },
       timer: null,
-      count: 0
+      count: 0,
+      loading: false
     }
   },
   computed: {
@@ -101,14 +94,23 @@ export default {
     // 先加载消息类型
     loadMsgTypes()
     this.queryCount()
+    // this.search()
     // 每30s定时刷新拉取消息
-    this.timer = setInterval(() => {
+    /* this.timer = setInterval(() => {
       this.queryCount()
-    }, 30000)
+    }, 30000) */
   },
   mounted() {
     bus.$on('refresh-top-notification', () => {
-      this.search()
+      if (this.visible) this.search()
+      else this.queryCount()
+    })
+
+    bus.$on('web-msg-topic-call', msg => {
+      console.log('收到websocket消息')
+      this.count = (this.count || this.paginationConfig.total) + 1
+      // this.queryCount()
+      // this.search()
     })
   },
   beforeDestroy() {
@@ -195,6 +197,7 @@ export default {
       })
     },
     search() {
+      this.loading = true
       const param = {
         status: false,
         orders: [' create_time desc ']
@@ -204,7 +207,9 @@ export default {
         this.data = response.data.listObject
         this.paginationConfig.total = response.data.itemCount
         this.count = this.paginationConfig.total
+        this.loading = false
       }).catch(() => {
+        this.loading = false
         const token = getToken()
         if (!token || token === 'null' || token === 'undefined') {
           this.timer && clearInterval(this.timer)

frontend/src/main.js

@@ -25,6 +25,7 @@ import '@/components/canvas/custom-component' // 注册自定义组件
 
 import '@/utils/DateUtil'
 import draggable from 'vuedraggable'
+import deWebsocket from '@/websocket'
 Vue.config.productionTip = false
 Vue.use(VueClipboard)
 Vue.use(widgets)
@@ -113,6 +114,7 @@ Vue.prototype.checkPermission = function(pers) {
   })
   return hasPermission
 }
+Vue.use(deWebsocket)
 new Vue({
 
   router,

frontend/src/permission.js

@@ -19,6 +19,8 @@ import {
 import Layout from '@/layout/index'
 // import bus from './utils/bus'
 
+import { getSocket } from '@/websocket'
+
 NProgress.configure({
   showSpinner: false
 }) // NProgress Configuration
@@ -57,6 +59,8 @@ router.beforeEach(async(to, from, next) => {
         if (store.getters.roles.length === 0) { // 判断当前用户是否已拉取完user_info信息
           // get user info
           store.dispatch('user/getInfo').then(() => {
+            const deWebsocket = getSocket()
+            deWebsocket && deWebsocket.reconnect && deWebsocket.reconnect()
             store.dispatch('lic/getLicInfo').then(() => {
               loadMenus(next, to)
             }).catch(() => {

frontend/src/websocket/index.js

+import bus from '@/utils/bus'
+import SockJS from 'sockjs-client'
+import Stomp from 'stompjs'
+import store from '@/store'
+
+class DeWebsocket {
+  constructor() {
+    this.ws_url = '/websocket'
+    this.client = null
+    this.channels = [
+      {
+        topic: '/web-msg-topic',
+        event: 'web-msg-topic-call'
+      }
+    ]
+    this.timer = null
+    this.initialize()
+  }
+
+  initialize() {
+    this.connection()
+    const _this = this
+    this.timer = this.isLoginStatu() && setInterval(() => {
+      this.isLoginStatu() || this.destroy()
+      try {
+        _this.client && _this.client.send('heart detection')
+      } catch (error) {
+        console.log('Disconnection reconnection...')
+        _this.connection()
+      }
+    }, 5000)
+  }
+
+  destroy() {
+    this.timer && clearInterval(this.timer)
+    this.disconnect()
+    return true
+  }
+
+  reconnect() {
+    this.initialize()
+  }
+
+  isLoginStatu() {
+    return store.state && store.state.user && store.state.user.user && store.state.user.user.userId
+  }
+
+  connection() {
+    const socket = new SockJS(this.ws_url)
+    /* const socket = new SockJS('http://localhost:8081' + this.ws_url) */
+    if (!this.isLoginStatu()) {
+      return
+    }
+    this.client = Stomp.over(socket)
+    const heads = {
+      /* Authorization: '', */
+      userId: store.state.user.user.userId
+    }
+
+    this.client.connect(
+      heads,
+      res => {
+        // 连接成功 订阅所有主题
+        this.subscribe()
+      },
+      err => {
+        // 连接失败 打印错误信息
+        console.error(err)
+      }
+    ).bind(this)
+  }
+  subscribe() {
+    this.channels.forEach(channel => {
+      this.client.subscribe('/user/' + store.state.user.user.userId + channel.topic, res => {
+        res && res.body && bus.$emit(channel.event, res.body)
+      })
+    })
+  }
+  disconnect() {
+    this.client && this.client.disconnect()
+  }
+}
+
+const result = new DeWebsocket()
+export const getSocket = () => {
+  return result
+}
+export default {
+  install(Vue) {
+    // 使用$$前缀，避免与Element UI的冲突
+    Vue.prototype.$deWebsocket = result
+  }
+}
+

frontend/vue.config.js

@@ -20,7 +20,7 @@ module.exports = {
     proxy: {
       '^(?!/login)': {
         target: 'http://localhost:8081/',
-        ws: false
+        ws: true
       }
     },
     open: true,