blog/src/services/socket.service.ts

import { injectable, Service } from './service'
import io from 'socket.io-client'
import TimerService from './timer.service'
import useStore from './user.data'

/**
 * socket io 返回数据格式
 */
type ResponseData = {
  status: number;
  body: object | null;
};

/**
 * 网络服务返回数据格式
 */
interface Result<T = Any> {
  success: boolean;
  data: T;
}

interface Request<T = Any> {
  code: string;
  timestamp: number;
  nonestr: string;
  body: T;
}

/**
 * 约束socket 请求响应参数，code声明，规范请求、返回参数
 */
interface SendKey<T = Record<string, Any>, P = Record<string, Any>> { }

/**
 * API网络请求服务
 */
@injectable
export default class SocketService extends Service {
  private URL = import.meta.env.VITE_SOCKET_URL

  private userData = useStore()

  private timerService = new TimerService()

  /**
     * 是否已经完成时间同步
     * 防止刷新之后因为时间同步问题导致超时
     */
  private timeSynced_ = false

  /**
     * 是否已经完成时间同步
     */
  public get timeSynced() {
    return this.timeSynced_
  }

  /**
   * 网络请求监听集合
   */
  private listeners: { [code: string]: [(response: object) => void] } = {}

  /**
     * 请求回调集合
     */
  private callbacks: { [key: string]: (response: ResponseData) => void } = {}

  /**
     * 服务器当前时间戳
     */
  private timestamp = Math.floor(Date.now() / 1000)

  /**
     * socketio对象
     */
  private socket!: SocketIOClient.Socket

  /**
     * 接口超时列表
     */
  private timeoutList: { [code: string]: number } = {
    'system.common': 60 * 1000
  }

  constructor() {
    super()
    // 建立连接
    this.reset()
    // 设置定时心跳
    this.timerService.on('10s', this.ping.bind(this))
    this.timerService.on('5m', this.syncTime.bind(this))
    this.timerService.on('5s', () => {
      this.timestamp += 5
    })
    this.aysncTime()
  }

  public async aysncTime() {
    while (true) {
      await (new Promise((res) => {
        setTimeout(res, 500)
      }))
      if (this.timeSynced_) {
        break
      }
      console.log('等待时间同步')
    }
  }

  /**
    * 全局发送消息
    * @param code 代码
    * @param message 信息
    * @param timeout 超时 默认6秒
    */
  public async send<R = Any, Q = Any>(code: string | SendKey<Q, R>, data?: Q, timeout = 1000 * 15): Promise<Result<R | null>> {
    if (this.timeoutList[code as string]) {
      timeout = this.timeoutList[code as string]
    }
    // if (!this.timeSynced_) throw 没同步成功会阻断请求
    return await this.request(code, data || {}, timeout)
  }

  /**
     * 全局监听socketio事件
     * @param code 命令
     * @param handle 处理接口
     */
  public on(code: string, handle: (response: Any) => void, unique = false) {
    if (unique) {
      this.listeners[code] = [ handle ]
    } else {
      this.listeners[code] = this.listeners[code] || []
      this.listeners[code].push(handle)
    }
    return true
  }

  /**
     * 取消全局监听socketio事件
     * @param code 命令 * 代表清除所有
     * @param handle 处理接口
     */
  public off(code: string, handle?: (response: object) => void) {
    if (code === '*') {
      this.listeners = {}
      return true
    }
    this.listeners[code] = this.listeners[code] || []
    const index = this.listeners[code].indexOf(handle!)
    if (index > -1) {
      this.listeners[code].splice(index, 1)
      return true
    }
    return false
  }

  /**
    * 设置身份令牌
    * @param token 令牌
    */
  public setToken(token: string) {
    this.userData.token = token
  }

  /**
     * 发起请求
     * @param code 命令码
     * @param data 数据
     */
  private request<T, P>(code: string | SendKey<T, P>, data: T | {}, timeout = 1000 * 6): Promise<Result> {
    const that = this
    const nonestr = `${Math.floor(Math.random() * 1000000)}`
    // proto 加密
    // 发送
    this.socket.emit(
      'request',
      JSON.stringify({
        code,
        body: data,
        timestamp: this.timestamp,
        nonestr,
        token: this.userData.token
      })
    )
    return new Promise((res, rej) => {
      const timer = setTimeout(() => {
        if (that.callbacks[nonestr]) {
          delete that.callbacks[nonestr]
        }
        throw '请求超时'
      }, timeout)
      that.callbacks[nonestr] = (response: ResponseData) => {
        clearTimeout(timer)
        delete that.callbacks[nonestr]
        const status = +response.status
        if (status !== 0) {
          if (status === 99) this.reset()
          if (status === 401) {
            this.userData.setToken('')
            throw 401
          }
          throw response.body
        } else {
          res({ success: true, data: response.body })
        }
      }
    })
  }

  /**
     * 心跳监测
     */
  private async ping() {
    if (!this.userData.token) {
      return
    }
    await this.request('user.heart', {})
  }

  /**
     * 时间同步
     */
  private async syncTime() {
    const { success, data } = await this.request('server.getNowTime', {})
    if (success && data) {
      this.timestamp = data
      this.timeSynced_ = true
      console.log('同步成功')
    }
  }

  /**
     * 接收socketio返回的消息
     * @param res 回复消息体
     */
  private onResponse(res: string) {
    let response = null
    try {
      response = JSON.parse(res)
    } catch (error) {
      throw `接口返回数据异常 ${res}`
    }
    if (response && this.callbacks[response.nonestr]) {
      this.callbacks[response.nonestr](response)
    }
  }

  /**
     * 接收socketio 的请求消息
     * @param request 回复消息体
     */
  private onRequest(request: string | Request) {
    let data!: Request
    // 解析数据
    if (typeof request === 'string') {
      const temp = JSON.parse(request)
      if (temp) data = temp
    } else {
      data = request
    }
    // 调用监听接口
    if (this.listeners[data.code]) {
      const errs: string[] = []
      for (const handle of this.listeners[data.code]) {
        try {
          handle(JSON.parse(JSON.stringify(data)))
        } catch (err) {
          errs.push((err as Any).stack)
        }
      }
      if (errs.length) {
        throw `${errs}`
      }
    }
  }

  /**
     * 重置连接
     */
  private reset() {
    if (this.socket) {
      this.socket.close()
    }
    try {
      if (process.env.NODE_ENV === 'development') {
        const opts = { transports: [ 'websocket' ] }
        this.URL.includes('https') && Object.assign(opts, { forceNew: true, path: '/wss' })
        this.socket = io(this.URL, opts)
      } else {
        this.socket = io({
          forceNew: true,
          transports: [ 'websocket' ],
          path: '/wss'
        })
      }
      this.socket.on('request', this.onRequest.bind(this))
      this.socket.on('response', this.onResponse.bind(this))
      setTimeout(this.ping.bind(this), 300)
      setTimeout(this.syncTime.bind(this), 100)
    } catch (err) {
      throw `${err}`
    }
  }
}