1. 新增消息管理类 2. socket消息类型js改为枚举类 3. 重连函数在使用时以参数传递进来

hcong

2024-11-27 6584cbef8a5829950c20b507e17445b97650e73f

1. 新增消息管理类 2. socket消息类型js改为枚举类 3. 重连函数在使用时以参数传递进来

 src/components/bg-task/FYBgTaskCard.vue

@@ -46,7 +46,8 @@
import { useFetchData } from '@/composables/fetchData';
import bgtaskApi from '@/api/fysp/bgtaskApi';
import { enumBgTask, BG_TASK_TYPE, BG_TASK_STATUS } from '@/enum/bgTask';
import eventBus from '@/socket/eventBus.js';
import { SOCKET_MESSAGE_TYPE } from '@/enum/socketMessage';
import MessageManager from '@/socket/MessageManager.js';

export default {
  setup() {
@@ -79,7 +80,7 @@
  },
  methods: {
    registerBgTaskMessage() {
      eventBus.register('background_task', (data) => {
      MessageManager.register(SOCKET_MESSAGE_TYPE.BACKGROUND_TASK.name, (data) => {
        this.refreshTaskById(data)
      })
    },

 src/enum/socketMessage.js

@@ -1,17 +1,29 @@
const messageTypes = [
  { id: 0, value: '0', label: '心跳机制', name: 'heartbeat' },
  { id: 1, value: '1', label: '后台任务', name: 'background_task' },
  { id: 2, value: '2', label: '业务日志', name: 'business_log' }
];
import { Enum } from './enum';
const SOCKET_MESSAGE_TYPE = Enum({
  BACKGROUND_TASK: {
    id: 1,
    value: '1',
    label: '后台任务',
    name: 'background_task'
  },
  BUSINESS_LOG: { id: 2, value: '2', label: '业务日志', name: 'business_log' }
});
/**
 * æ ¹æ®socket消息类型字段得到这个类型对象
 * @param {*} value socket消息类型
 * @returns å¯¹åº”的消息类型对象
 */
/**
 * é€šè¿‡åç§°æŸ¥æ‰¾æžšä¸¾ç±»
 * @param {String} value
 * @returns
 */
function getByValue(value) {
  return messageTypes.find((v) => v.value == value);
  for (const type in SOCKET_MESSAGE_TYPE) {
    const typeObj = SOCKET_MESSAGE_TYPE[type]
    if (typeObj.value == value) {
      return typeObj
    }
  }
}
function isHeartbeatMessageByType(type) {
  return type == '0';
}
export { messageTypes, getByValue, isHeartbeatMessageByType };
export { SOCKET_MESSAGE_TYPE, getByValue };

 src/socket/FYWebSocket.js

@@ -1,9 +1,4 @@
import eventBus from './eventBus.js';
import {
  messageTypes,
  getByValue,
  isHeartbeatMessageByType
} from '@/enum/socketMessage.js';
import { SOCKET_MESSAGE_TYPE, getByValue } from '@/enum/socketMessage.js';
import { encodeMessage, decodeMessage } from './socketMessage.js';
import { $socket_base_url } from '@/api/index.js';
class FYWebSocket extends WebSocket {
@@ -16,15 +11,22 @@
   *    time: å¿ƒè·³æ—¶é—´é—´éš”
   *    timeout: å¿ƒè·³è¶…时间隔
   *    reconnect: æ–­çº¿é‡è¿žæ—¶é—´é—´éš”
   * isReconnect æ˜¯å¦æ–­çº¿é‡è¿ž
   * reconnectFunc æ–­çº¿é‡è¿žå‡½æ•°
   * messageManager æ¶ˆæ¯ç®¡ç†å™¨
   */
  init(heartBeatConfig, isReconnect) {
  init(reconnectFunc, messageManager) {
    this.onopen = this.openHandler; // è¿žæŽ¥æˆåŠŸåŽçš„回调函数
    this.onclose = this.closeHandler; // è¿žæŽ¥å…³é—­åŽçš„回调 å‡½æ•°
    this.onmessage = this.messageHandler; // æ”¶åˆ°æœåŠ¡å™¨æ•°æ®åŽçš„回调函数
    this.onerror = this.errorHandler; // è¿žæŽ¥å‘生错误的回调方法
    this.heartBeatConfig = heartBeatConfig; // å¿ƒè·³è¿žæŽ¥é…ç½®å‚æ•°
    this.isReconnect = isReconnect; // è®°å½•æ˜¯å¦æ–­çº¿é‡è¿ž
    this.heartBeatConfig = {
      time: 30 * 1000,
      timeout: 2 * 1000,
      reconnect: 30 * 1000
    }; // å¿ƒè·³è¿žæŽ¥é…ç½®å‚æ•°
    this.isReconnect = typeof reconnectFunc === 'function'; // è®°å½•æ˜¯å¦æ–­çº¿é‡è¿ž
    this.reconnectFunc = reconnectFunc; // æ–­çº¿é‡è¿žå‡½æ•°
    this.messageManager = messageManager; // æ¶ˆæ¯å¤„理管理器
    this.reconnectTimer = null; // è®°å½•æ–­çº¿é‡è¿žçš„时间器
    this.startHeartBeatTimer = null; // è®°å½•å¿ƒè·³æ—¶é—´å™¨
    this.webSocketState = false; // è®°å½•socket连接状态 true为已连接
@@ -39,7 +41,7 @@
  }
  // è¿žæŽ¥æˆåŠŸåŽçš„回调函数
  openHandler() {
    console.log('====onopen è¿žæŽ¥æˆåŠŸ====');
    console.log('====onopen websocket连接成功====');
    // socket状态设置为连接，做为后面的断线重连的拦截器
    this.webSocketState = true;
    // åˆ¤æ–­æ˜¯å¦å¯åŠ¨å¿ƒè·³æœºåˆ¶
@@ -53,22 +55,17 @@
    if (!(webSocketMsg && webSocketMsg != null && webSocketMsg != {})) {
      return;
    }
    console.log("webSocketMsg", webSocketMsg);
    
    const type = webSocketMsg.type;
    const data = webSocketMsg.data;
    let typeObj = getByValue(type);
    console.log("typeObj", typeObj);
    
    if (isHeartbeatMessageByType(type)) {
    if (type == 0) {
      // å°†è¿žæŽ¥çŠ¶æ€æ›´æ–°ä¸ºåœ¨çº¿
      this.webSocketState = true;
      eventBus.emit(typeObj.name, data);
      console.log('====onmessage å¿ƒè·³æ£€æµ‹====', data);
      console.log('====onmessage websocket心跳检测====', data);
    } else {
      // å‘送事件
      eventBus.emit(typeObj.name, data);
      console.log(`====onmessage ${typeObj.label}====`, data);
      this.messageManager.emit(typeObj.name, data);
      console.log(`====onmessage websocket${typeObj.label}====`, data);
    }
  }
  // è¿žæŽ¥å…³é—­åŽçš„回调 å‡½æ•°
@@ -93,7 +90,7 @@
  startHeartBeat(time) {
    this.startHeartBeatTimer = setTimeout(() => {
      // å®¢æˆ·ç«¯æ¯éš”一段时间向服务端发送一个心跳消息
      this.sendMessage(messageTypes[0].value, Date.now());
      this.sendMessage('0', Date.now());
      this.waitingServer();
    }, time);
  }
@@ -106,12 +103,12 @@
        this.startHeartBeat(this.heartBeatConfig.time);
        return;
      }
      console.log('心跳无响应, å·²ç»å’ŒæœåŠ¡ç«¯æ–­çº¿');
      console.log('websocket å¿ƒè·³æ— å“åº”, å·²ç»å’ŒæœåŠ¡ç«¯æ–­çº¿');
      // é‡æ–°è¿žæŽ¥æ—¶ï¼Œè®°å¾—要先关闭当前连接
      try {
        this.close();
      } catch (error) {
        console.log('当前连接已经关闭');
        console.log('websocket å½“前连接已经关闭');
      }
      // // é‡æ–°è¿žæŽ¥
      // this.reconnectWebSocket()
@@ -126,8 +123,9 @@
    }
    // æ ¹æ®ä¼ å…¥çš„断线重连时间间隔 å»¶æ—¶è¿žæŽ¥
    this.reconnectTimer = setTimeout(() => {
      // è§¦å‘重新连接事件
      eventBus.emit('reconnect');
      // è§¦å‘重新连接函数
      console.log('====onerror websocket尝试重连====');
      this.reconnectFunc();
    }, this.heartBeatConfig.reconnect);
  }
}

 src/socket/MessageManager.js

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,27 @@
import eventBus from './eventBus.js';

class MessageManager {
  constructor() {
    this.messageHandler = eventBus;
  }
  /**
   * æ³¨å†Œäº‹ä»¶
   * @param {*} eventType äº‹ä»¶ç±»åž‹
   * @param {*} handler äº‹ä»¶å¤„理函数
   */
  register(eventType, handler) {
    this.messageHandler.register(eventType, handler);
  }

  /**
   * å‘送事件
   * @param {*} eventType äº‹ä»¶ç±»åž‹
   * @param  {...any} ars äº‹ä»¶å†…容
   */
  emit(eventType, ...ars) {
    this.messageHandler.emit(eventType, ...ars);
  }
}
const instance = new MessageManager();
Object.freeze(instance); // é˜²æ­¢ä¿®æ”¹å®žä¾‹
export default instance;

 src/socket/index.js

@@ -1,31 +1,20 @@

import { FYWebSocket } from '@/socket/FYWebSocket.js'
import eventBus from '@/socket/eventBus.js';
import { FYWebSocket } from '@/socket/FYWebSocket.js';
import MessageManager from '@/socket/MessageManager.js';

let socket = null;

// è¿žæŽ¥websocket
function connectWebSocket() {
  if (socket == null) {
    socket = new FYWebSocket()
  }
  socket.init({
    time: 4 * 1000,
    timeout: 2 * 1000,
    reconnect: 3 * 1000
  }, true)
  socket = new FYWebSocket();
  socket.init(() => {
    connectWebSocket();
  }, MessageManager);
  return socket;
}
// æ–­çº¿é‡è¿žwebsocket
function startReconnectWebSocket() {
  eventBus.register('reconnect', () => {
    connectWebSocket()
  })
}
/**
 * åˆå§‹åŒ–socket客户端
 */
function initSocketClient() {
  connectWebSocket();
  startReconnectWebSocket();
}
export { initSocketClient }
export { initSocketClient };

 src/socket/socketMessage.js

@@ -1,4 +1,3 @@
import { messageTypes } from '@/enum/socketMessage.js' 
// å¼€å§‹ç¬¦å·å’Œç»“束符号分别为 '##' å’Œ '%%', åˆ†éš”符为 &&
// å¼€å§‹ç¬¦å·
const startStr = '##';
@@ -35,7 +34,7 @@
  const type = parts[0];
  let data = JSON.parse(parts[1]);
  return {
    type: messageTypes.find((item) => item.value == type).value,
    type: type,
    data: data
  };
}