fe.ems.vue3/src/views/dashboard/overview/hooks/useWS.ts

import { RESULT_CODE_ERROR } from '@/constants/result-constants';
import { OptionsType, WS } from '@/plugins/ws-websocket';
import { onBeforeUnmount, onMounted } from 'vue';
import {
  ueEventParse,
  cdrEventParse,
  eventData,
  eventTotal,
  eventId,
  userActivityReset,
} from './useUserActivity';
import {
  upfTotalFlow,
  upfTFParse,
  upfFlowParse,
  upfTotalFlowReset,
} from './useUPFTotalFlow';
import { topologyReset, neStateParse } from './useTopology';
import PQueue from 'p-queue';

/**websocket连接 */
export default function useWS() {
  const ws = new WS();
  const queue = new PQueue({ concurrency: 1, autoStart: true });

  /**发消息 */
  function wsSend(data: Record<string, any>) {
    ws.send(data);
  }

  /**接收数据后回调 */
  function wsError(ev: any) {
    // 接收数据后回调
    console.error(ev);
  }

  /**接收数据后回调 */
  function wsMessage(res: Record<string, any>) {
    // console.log(res);
    const { code, requestId, data } = res;
    if (code === RESULT_CODE_ERROR) {
      console.warn(res.msg);
      return;
    }

    // 网元状态
    if (requestId && requestId.startsWith('neState')) {
      const neType = requestId.split('_')[1];
      neStateParse(neType, data);
      return;
    }

    // 普通信息
    switch (requestId) {
      // ueEvent UE会话事件
      case '1010':
        if (Array.isArray(data.rows)) {
          eventTotal.value += data.total;
          for (const item of data.rows) {
            const v = ueEventParse(item);
            if (v) {
              eventData.value.push(v);
            }
          }
          if (eventData.value.length > 0) {
            eventId.value = eventData.value[0].eId;
          }
        }
        break;
      //cdrEvent CDR会话事件
      case '1005':
        if (Array.isArray(data.rows)) {
          eventTotal.value += data.total;
          for (const item of data.rows) {
            const v = cdrEventParse(item);
            if (v) {
              eventData.value.push(v);
            }
          }
          if (eventData.value.length > 0) {
            eventId.value = eventData.value[0].eId;
          }
        }
        break;
      //UPF-总流量数
      case '1030_0':
        const v0 = upfTFParse(data);
        upfTotalFlow.value[0].up = v0.up;
        upfTotalFlow.value[0].down = v0.down;
        upfTotalFlow.value[0].requestFlag = false;
        break;
      case '1030_7':
        const v7 = upfTFParse(data);
        upfTotalFlow.value[1].up = v7.up;
        upfTotalFlow.value[1].down = v7.down;
        upfTotalFlow.value[1].requestFlag = false;
        break;
      case '1030_30':
        const v30 = upfTFParse(data);
        upfTotalFlow.value[2].up = v30.up;
        upfTotalFlow.value[2].down = v30.down;
        upfTotalFlow.value[2].requestFlag = false;
        break;
    }

    // 订阅组信息
    if (!data?.groupId) {
      return;
    }
    switch (data.groupId) {
      // kpiEvent 指标UPF
      case '12':
        if (data.data) {
          upfFlowParse(data.data);
        }
        break;
      // ueEvent UE会话事件
      case '1010':
        if (data.data) {
          queue.add(async () => {
            const v = ueEventParse(data.data);
            if (v) {
              eventData.value.unshift(v);
              eventTotal.value += 1;
              eventId.value = v.eId;
              await new Promise(resolve => setTimeout(resolve, 800));
              if (eventData.value.length > 20) {
                eventData.value.pop();
              }
            }
          });
        }
        break;
      // cdrEvent CDR会话事件
      case '1005':
        if (data.data) {
          queue.add(async () => {
            const v = cdrEventParse(data.data);
            if (v) {
              eventData.value.unshift(v);
              eventTotal.value += 1;
              eventId.value = v.eId;
              await new Promise(resolve => setTimeout(resolve, 800));
              if (eventData.value.length > 20) {
                eventData.value.pop();
              }
            }
          });
        }
        break;
    }
  }

  /**UPF-总流量数 发消息*/
  function upfTFSend(day: 0 | 7 | 30) {
    // 请求标记检查避免重复发送
    let index = 0;
    if (day === 0) {
      index = 0;
    } else if (day === 7) {
      index = 1;
    } else if (day === 30) {
      index = 2;
    }
    if (upfTotalFlow.value[index].requestFlag) {
      return;
    }
    upfTotalFlow.value[index].requestFlag = true;

    ws.send({
      requestId: `1030_${day}`,
      type: 'upf_tf',
      data: {
        neType: 'UPF',
        neId: '001',
        day: day,
      },
    });
  }

  /**ueEvent UE会话事件 发消息*/
  function ueEventSend() {
    ws.send({
      requestId: '1010',
      type: 'ue',
      data: {
        neType: 'AMF',
        neId: '001',
        sortField: 'timestamp',
        sortOrder: 'desc',
        pageNum: 1,
        pageSize: 10,
      },
    });
  }

  /**cdrEvent CDR会话事件 发消息*/
  function cdrEventSend() {
    ws.send({
      requestId: '1005',
      type: 'cdr',
      data: {
        neType: 'IMS',
        neId: '001',
        recordType: 'MOC',
        sortField: 'timestamp',
        sortOrder: 'desc',
        pageNum: 1,
        pageSize: 10,
      },
    });
  }

  onMounted(() => {
    const options: OptionsType = {
      url: '/ws',
      params: {
        /**订阅通道组
         *
         * 指标UPF (GroupID:12)
         * UE会话事件-AMF (GroupID:1010)
         * CDR会话事件-IMS (GroupID:1005)
         */
        subGroupID: '12,1010,1005',
      },
      onmessage: wsMessage,
      onerror: wsError,
    };
    ws.connect(options);
  });

  onBeforeUnmount(() => {
    ws.close();
    userActivityReset();
    upfTotalFlowReset();
    topologyReset();
  });

  return {
    wsSend,
    cdrEventSend,
    ueEventSend,
    upfTFSend,
  };
}