wxappb2b/GuangZuan/miniprogram/utils/websocket/index.ts

import { Package40, Package50 } from './package';
import { Package, ConnectionState, SendMessage, AsyncTask } from './interface';
import { v4 } from '../uuid/index'
import moment from 'moment'

/**
 * MTP2.0 长链通信类
 */
export class MTP2WebSocket<T extends Package40 | Package50>{
  /** 报文协议 */
  private readonly Package: Package<T>;
  /** 当前实例标识 */
  private uuid = '';
  /** 服务端地址 */
  private host;
  /** WebSocket 对象 */
  private ws?: WechatMiniprogram.SocketTask;
  /** 心跳定时器 */
  private beatTimer = 0;
  /** 心跳发送时间 */
  private beatTimerId = '';
  /** 重连定时器 */
  private reconnectTimer = 0;
  /** 超时定时器 */
  private timeoutTimer = 0;
  /** 当前流水号 */
  private currentSerial = 1;
  /** 信息发送异步建值对 */
  private asyncTaskMap = new Map<string, AsyncTask<T>>();
  /** 初始默认重连间隔时长 */
  private defaultReconnectInterval = 5 * 1000;
  /** 连接准备完成 */
  private onReady?: Promise<MTP2WebSocket<T>>;

  /** 连接成功的回调 */
  onOpen?: (socket: MTP2WebSocket<T>) => void;
  /** 连接断开的回调 */
  onClose?: () => void;
  /** 连接发生错误的回调 */
  onError?: (e: string) => void;
  /** 接收到推送类报文的回调 */
  onPush?: (data: T) => void;
  /** 重连状态发生改变时的回调 */
  onReconnect?: (socket: MTP2WebSocket<T>) => void;

  /** 当前连接状态 */
  connState: keyof typeof ConnectionState = 'Unconnected';
  /** 重连次数 */
  reconnectCount = 0;
  /** 心跳间隔时长，默认为30秒 */
  beatInterval = 30 * 1000;
  /** 重连间隔时长，默认为5秒 */
  reconnectInterval = this.defaultReconnectInterval;
  /** 消息超时时长，默认为15秒 */
  timeoutInterval = 15 * 1000;

  constructor(protocol: Package<T>, host = '') {
    this.Package = protocol;
    this.host = host;
  }

  /**
   * 连接服务器
   */
  connect(host?: string): Promise<MTP2WebSocket<T>> {
    if (!this.onReady) {
      clearTimeout(this.reconnectTimer);
      this.stopHeartBeat();
      this.host = host || this.host;
      this.connState = 'Connecting';
      console.log(this.Package.name, this.host, '正在连接');

      this.onReady = new Promise((resolve, reject) => {
        const uuid = v4();
        this.uuid = uuid;
        this.ws = wx.connectSocket({ url: this.host });

        try {
          // 连接成功
          this.ws.onOpen(() => {
            console.log(this.Package.name, this.host, '连接成功');
            this.connState = 'Connected';
            this.startHeartBeat();
            this.onOpen && this.onOpen(this);
            resolve(this);
          })
          // 连接断开(CLOSING有可能延迟)
          this.ws.onClose(() => {
            // 判断是否当前实例
            // 如果连接断开后不等待 onclose 响应，由于 onclose 延迟的原因可能会在创建新的 ws 实例后触发，导致刚创建的实例被断开进入重连机制
            if (this.uuid === uuid) {
              console.warn(this.Package.name, this.host, '连接已断开');
              this.reset();
              this.reconnect();
            }
          })
          // 连接发生错误
          this.ws.onError((e) => {
            const message = this.host + '连接发生错误';
            this.onError && this.onError(e.errMsg);
            reject(message);
          })
          // 接收数据
          this.ws.onMessage((e) => {
            if (e.data instanceof ArrayBuffer) {
              this.disposeReceiveDatas(new Uint8Array(e.data));
            } else {
              const data = e.data.split('').map((val, index) => val.charCodeAt(index));
              this.disposeReceiveDatas(new Uint8Array(data));
            }
          })
        }
        catch {
          reject(this.Package.name + ' 连接失败');
        }
      })
    }
    return this.onReady;
  }

  /**
   * 主动断开连接，断开后不会自动重连
   */
  close() {
    clearTimeout(this.reconnectTimer);
    this.stopHeartBeat();
    this.uuid = v4();
    this.reconnectCount = 0;
    this.reconnectInterval = this.defaultReconnectInterval;

    if (this.ws) {
      this.ws.close({ reason: '主动断开' });
      console.warn(this.Package.name, this.host, '主动断开');
    }

    this.reset(); // 不等待 ws.onclose 响应强制重置实例
  }

  /**
   * 重置实例
   */
  private reset() {
    this.ws = undefined;
    this.onReady = undefined;
    this.connState = 'Unconnected';
    this.onClose && this.onClose();
  }

  /**
  * 启动心跳请求包发送Timer
  */
  private startHeartBeat() {
    this.beatTimer = setTimeout(() => {
      this.beatTimerId = moment(new Date()).format('HH:mm:ss');
      console.log(this.Package.name, '发送心跳', this.beatTimerId);
      // 发送心跳
      switch (this.Package) {
        case Package40: {
          // 4.0
          this.send({
            data: { rspCode: -1, payload: new this.Package(0x12) }
          })
          break;
        }
        case Package50: {
          // 5.0
          this.send({
            data: { payload: new this.Package(0) }
          })
          break;
        }
      }
      // 如果已经超过或心跳超时时长没有收到心跳回复，则认为网络已经异常，进行断网重连
      // WARNING: - 注意在正常断网时停止心跳逻辑
      this.timeoutTimer = setTimeout(() => {
        console.warn(this.Package.name, '心跳超时', this.beatTimerId);
        this.reconnect();
      }, this.timeoutInterval)
    }, this.beatInterval)
  }

  /**
   * 停止心跳请求包发送Timer
   */
  private stopHeartBeat() {
    clearTimeout(this.timeoutTimer);
    clearTimeout(this.beatTimer);
  }

  /**
   * 断网重连方法，在重连尝试失败时会递归调用自己
   */
  private reconnect() {
    this.stopHeartBeat();
    if (this.connState !== 'Connecting') {
      this.reconnectCount++;
      console.log(this.Package.name, this.host, `${this.reconnectInterval / 1000}秒后将进行第${this.reconnectCount}次重连`);

      this.reconnectTimer = setTimeout(() => {
        this.onReady = undefined;
        this.connect().then(() => {
          console.log(this.Package.name, this.host, '重连成功，可开始进行业务操作');
          this.reconnectCount = 0;
          this.reconnectInterval = this.defaultReconnectInterval;
          this.onReconnect && this.onReconnect(this);
        }).catch(() => {
          if (this.reconnectCount) {
            this.reconnectInterval += this.defaultReconnectInterval; // 重连间隔时间每次递增5秒
            console.warn(this.Package.name, this.host, `第${this.reconnectCount}次重连失败`);
          }
        })
      }, this.reconnectInterval);
    }
  }

  /**
   * 发送报文
   * @param data 目标报文
   * @param fncode 回复号。P为Package40对象时为回复大类号，同时必传；P为Package50对象为回复功能码，可不传。
   * @param success 成功回调
   * @param fail 失败回调
   */
  send(msg: SendMessage<T>) {
    const { data, success, fail } = msg;
    const { rspCode, payload } = data;

    if (this.onReady) {
      this.onReady.then(() => {
        // 设置流水号
        payload.serialNumber = this.currentSerial;
        this.currentSerial++;

        let key = payload.serialNumber.toString();
        // 4.0报文
        if (payload instanceof Package40) {
          if (!rspCode && fail) {
            fail('缺少rspCode参数');
            return;
          }
          key += '_' + rspCode;
        }

        // 判断是否需要异步回调
        if (success || fail) {
          // 保存发送任务，为任务设置一个自动超时定时器
          this.asyncTaskMap.set(key, {
            sendMessage: msg,
            timeoutId: setTimeout(() => this.asyncTimeout(key), this.timeoutInterval),
          })
        }

        // 发送信息
        this.ws?.send({ data: payload.data().buffer });
      })
    } else {
      fail && fail('服务未连接');
    }
  }

  /**
   * 异步任务超时
   * @param key key
   */
  private asyncTimeout(key: string) {
    // 获取对应异步任务对象
    const asyncTask = this.asyncTaskMap.get(key);
    const onFail = asyncTask?.sendMessage.fail;
    onFail && onFail('业务超时');
    this.asyncTaskMap.delete(key);
  }

  /**
   * 处理接收数据
   * @param bytes 接收的数据
   */
  private disposeReceiveDatas(bytes: Uint8Array) {
    const cache: number[] = [];
    let cachePackageLength = 0;

    for (let i = 0; i < bytes.length; i++) {
      const byte = bytes[i];
      cache.push(byte);

      if (i === 0 && byte !== 0xff) {
        console.error('接收到首字节不是0xFF');
        return;
      } else {
        // 取报文长度
        if (cache.length === 5) {
          switch (this.Package) {
            // 4.0报文
            case Package40: {
              cachePackageLength = new DataView(new Uint8Array(new Uint8Array(cache).subarray(1, 5)).buffer).getUint32(0, false);
              if (cachePackageLength > 65535) {
                console.error('接收到长度超过65535的行情包');
                return;
              }
              break;
            }
            // 5.0报文
            case Package50: {
              const sub = new Uint8Array(cache).subarray(1, 5);
              const dataView = new DataView(new Uint8Array(sub).buffer); // 注意这里要new一个新的Uint8Array，直接buffer取到的还是原数组的buffer
              cachePackageLength = dataView.getUint32(0, true);
              break;
            }
          }
        }
        // 判断是否已经到包尾
        if (cache.length === cachePackageLength) {
          if (byte !== 0x0) {
            console.error('接收到尾字节不是0x0的错误数据包');
            return;
          }

          const content = new Uint8Array(cache);
          const result = new this.Package(content);

          if (result.packageLength === 0) {
            console.error('报文装箱失败');
            return;
          }
          this.disposePackageResult(result);
        }
      }
    }
  }

  /**
   * 处理完整的报文
   */
  private disposePackageResult(p: T) {
    let key = p.serialNumber.toString();
    if (p instanceof Package40) {
      key += '_' + p.mainClassNumber;
    }
    const asyncTask = this.asyncTaskMap.get(key);
    const sendMessage = asyncTask?.sendMessage;

    // 移除当前发送的任务请求
    if (asyncTask) {
      clearTimeout(asyncTask.timeoutId);
      this.asyncTaskMap.delete(key);
    }

    // 4.0报文
    if (p instanceof Package40) {
      // 推送类报文, 0x12 - 心跳, 0x41 - 实时行情推送, 0x42 - 控制信号
      switch (p.mainClassNumber) {
        case 0x12: {
          // 接收到心跳回复
          console.log(this.Package.name, '收到心跳回复', this.beatTimerId);
          this.stopHeartBeat();
          this.startHeartBeat();
          break;
        }
        case 0x41:
        case 0x42: {
          // 推送类报文
          this.onPush && this.onPush(p);
          break;
        }
        default: {
          // 非推送类报文
          sendMessage?.success && sendMessage.success(p);
        }
      }
    }

    // 5.0报文
    if (p instanceof Package50) {
      if (p.funCode === 0) {
        // 接收到心跳回复
        console.log(this.Package.name, '收到心跳回复', this.beatTimerId);
        this.stopHeartBeat();
        this.startHeartBeat();
      } else if (p.serialNumber === 0) {
        // 推送类报文
        this.onPush && this.onPush(p);
      } else if (sendMessage) {
        // 非推送类报文
        const { data, success, fail } = sendMessage;
        // 判断是否需要检验回复功能码
        if (data.rspCode && data.rspCode !== p.funCode) {
          fail && fail('无效的rspCode：');
        } else {
          success && success(p);
        }
      }
    }
  }
}