wx-sse-client
遇到的问题：

文字丢失问题

由于是收到的是数据流，前端接收到的数据块，在解析的时候发现有些是不完整的，无法解析。反应到界面上就是发现会丢失一些文字。

解决方案：保存未处理的数据，新的数据来的时候，拼接上新数据，尝试解析，解析后，保留未能解析的数据，给一下段数据使用

/**
 * SSE 客户端封装类
 * 支持自动重连、错误处理和事件监听
 * events: open, message, readystatechange, error, close, success, complete, reconnecting
 */

class SSECustomEvent {
  constructor(type, options) {
    this.type = type
    // this.responseCode = options.responseCode
    // this.headers = options.headers
  }
}

/**
 * SSE 客户端封装类
 * 支持自动重连、错误处理和事件监听
 * 参考：https://github.com/mpetazzoni/sse.js
 */
class SSEClient {
  /** @type {number} */
  /**
   * -1 初始化中
   */
  static INITIALIZING = -1
  /** @type {number} */
  /**
   * 0 连接中
   */
  static CONNECTING = 0
  /** @type {number} */
  /**
   * 1 已连接
   */
  static OPEN = 1
  /** @type {number} */
  /**
   * 2 已关闭
   */
  static CLOSED = 2

  constructor(url, options = {}) {
    options = options || {}

    // 配置项初始化
    this.url = url || '' // SSE 服务器地址
    this.reconnectInterval = options.reconnectInterval || 10000 // 重连间隔时间(毫秒)
    this.maxRetries = options.maxRetries || 5 // 最大重试次数

    this.headers = options.headers || {} // 请求头
    this.payload = options.payload !== undefined ? options.payload : '' // 请求体
    this.method = options.method || (this.payload && 'POST') || 'GET' // 请求方法
    // this.withCredentials = !!options.withCredentials
    this.debug = !!options.debug // 是否开启调试
    // this.debug = true
    // 内部状态
    this.retryCount = 0 // 当前重试次数
    this.requestTask = null // 请求任务对象
    this.listeners = {} // 事件监听器集合

    /** @type {string} */
    this.FIELD_SEPARATOR = ':' // 字段分隔符

    /** @type {number} */
    this.readyState = SSEClient.INITIALIZING // 连接状态
    // /** @type {number} */
    // this.progress = 0
    /** @type {string} */
    this.chunk = ''
    /** @type {string} */
    this.lastEventId = ''
  }

  /**
   * 建立连接
   */
  connect() {
    // 如果连接状态为连接中或已连接，则直接返回
    if (
      this.readyState === SSEClient.CONNECTING ||
      this.readyState === SSEClient.OPEN
    )
      return

    this._onConnecting()

    this.requestTask = wx.request({
      url: this.url,
      method: this.method,
      header: {
        ...this.headers,
        Accept: 'text/event-stream',
        'Cache-Control': 'no-cache'
      },
      enableChunked: true, // 开启分块传输
      success: (res) => {
        // 断开的时候回调
        if (this.debug) {
          console.debug('SSE success', res)
        }
        this._onSuccess(res)
        // // todo 处理响应码非200的错误
        // if (res.statusCode !== 200) {
        //   this._onFailure(new Error(`SSE 连接失败，响应码：${res.statusCode}`))
        //   return
        // }

        // this._onOpened()
      },
      fail: (error) => {
        if (this.debug) {
          console.debug('SSE fail', error)
        }
        this._onFailure(error)
        this._reconnect()
      },
      complete: () => {
        if (this.debug) {
          console.debug('SSE complete')
        }
        this._onComplete()
      }
    })

    console.log('requestTask:', this.requestTask)

    // 监听数据返回
    this.requestTask.onChunkReceived((response) => {
      try {
        this._onOpened()
        // if (this.debug) {
        //   console.debug('SSE onChunkReceived', this.requestTask, response)
        // }
        // 将 ArrayBuffer 转换为字符串
        const data = this._arrayBufferToString(response.data)
        // 将数据分隔为多个部分
        const parts = (this.chunk + data).split(/(\r\n\r\n|\r\r|\n\n)/g)

        /*
         * We assume that the last chunk can be incomplete because of buffering or other network effects,
         * so we always save the last part to merge it with the next incoming packet
         */
        const lastPart = parts.pop()
        // 遍历每个部分，解析事件
        parts.forEach((part) => {
          if (part.trim().length > 0) {
            if (this.debug) {
              console.debug('SSE onChunkReceived part', part)
            }
            this._dispatchEvent(this._parseEventChunk(part))
          }
        })
        this.chunk = lastPart
      } catch (error) {
        console.error('SSE onChunkReceived error', error)
      }
    })
  }

  /**
   * 关闭连接
   */
  close() {
    if (this.readyState === SSEClient.CLOSED) {
      return
    }
    if (this.requestTask) {
      this.requestTask.abort()
    }
    this._markClosed()
  }

  /**
   * 添加事件监听器
   * @param {string} event 事件名称 (message/open/close/error)
   * @param {Function} callback 回调函数
   */
  on(event, callback) {
    if (!this.listeners[event]) {
      this.listeners[event] = []
    }
    const callbacks = this.listeners[event]
    if (callbacks.indexOf(callback) === -1) {
      callbacks.push(callback)
    }
  }

  /**
   * 移除事件监听器
   * @param {string} event 事件名称
   * @param {Function} callback 回调函数
   */
  off(event, callback) {
    if (this.listeners[event]) {
      if (!callback) {
        // 移除所有事件监听器
        this.listeners[event] = []
      } else {
        const callbacks = this.listeners[event]
        const newCallbacks = callbacks.filter((cb) => cb !== callback)
        // 如果移除后没有监听器，则删除事件
        if (newCallbacks.length === 0) {
          delete this.listeners[event]
        } else {
          this.listeners[event] = newCallbacks
        }
      }
    }
  }

  /**
   * 触发事件
   * @param {SSECustomEvent} e 事件对象
   * @returns {boolean} 是否成功触发事件
   * @private
   */
  _dispatchEvent(e) {
    if (!e) {
      return true
    }

    const type = e.type
    // 设置事件源
    e.source = this

    if (this.listeners[type]) {
      // 遍历所有事件监听器，并执行回调函数
      this.listeners[type].every((callback) => {
        callback(e)
        // 如果事件被阻止，则停止遍历
        return !e.defaultPrevented
      })
    }
    return true
  }

  /**
   * 标记连接已关闭
   * @private
   */
  _markClosed() {
    // this.chunk = ''
    // this.requestTask = null
    this._setReadyState(SSEClient.CLOSED)
  }

  /**
   * 设置连接状态
   * @param {number} state 连接状态
   * @private
   */
  _setReadyState(state) {
    if (this.readyState === state) {
      return
    }
    if (this.debug) {
      console.debug('SSE _setReadyState', state, this._stateToString(state))
    }
    const event = new SSECustomEvent('readystatechange')
    event.readyState = state
    this.readyState = state
    this._dispatchEvent(event)
  }

  /**
   * 将状态转换为字符串
   * @param {number} state 状态
   * @returns {string} 状态字符串
   */
  _stateToString(state) {
    return ['初始化中', '连接中', '已连接', '已关闭'][state + 1]
  }

  /**
   * 标记连接已打开
   * @private
   */
  _onOpened() {
    if (this.readyState === SSEClient.OPEN) {
      return
    }
    this.chunk = ''
    this.retryCount = 0
    // 触发 open 事件
    const event = new SSECustomEvent('open')
    event.responseCode = this.requestTask.status
    event.headers = this.headers
    this._setReadyState(SSEClient.OPEN)
    this._dispatchEvent(event)
  }

  _onConnecting() {
    this._setReadyState(SSEClient.CONNECTING)
    const event = new SSECustomEvent('connecting')
    this._dispatchEvent(event)
  }

  /**
   * 处理失败
   * @param {Error} e 错误对象
   * @private
   */
  _onFailure(e) {
    const event = new SSECustomEvent('error')
    event.error = e
    this._dispatchEvent(event)
    this._markClosed()
  }

  /**
   * 处理成功
   * @param {Object} e 成功对象
   * @private
   */
  _onSuccess(e) {
    const event = new SSECustomEvent('success')
    this._dispatchEvent(event)
    this._markClosed()
  }

  /**
   * 处理完成
   * @param {*} e 完成对象
   * @private
   */
  _onComplete(e) {
    const event = new SSECustomEvent('complete')
    this._dispatchEvent(event)
    this._markClosed()
  }

  /**
   * 重连机制
   * @private
   */
  _reconnect() {
    if (this.retryCount >= this.maxRetries) {
      const event = new SSECustomEvent('error')
      event.error = new Error('最大重试次数已达到')
      this._dispatchEvent(event)
      return
    }

    this.retryCount++
    setTimeout(() => {
      const event = new SSECustomEvent('reconnecting')
      event.retryCount = this.retryCount
      this._dispatchEvent(event)
      this.connect()
    }, this.reconnectInterval)
  }

  /**
   * 将 ArrayBuffer 转换为字符串
   * @private
   */
  _arrayBufferToString(buffer) {
    let text = decodeURIComponent(
      escape(String.fromCharCode.apply(null, new Uint8Array(buffer)))
    )
    return text
  }

  /**
   * 解析接收到的 SSE 事件块，构造事件对象
   * Parse a received SSE event chunk into a constructed event object.
   *
   * Reference: https://html.spec.whatwg.org/multipage/server-sent-events.html#dispatchMessage
   */
  _parseEventChunk(chunk) {
    if (!chunk || chunk.length === 0) {
      return null
    }

    const e = { id: null, retry: null, data: null, event: null }
    chunk.split(/\n|\r\n|\r/).forEach((line) => {
      const index = line.indexOf(this.FIELD_SEPARATOR)
      let field, value
      if (index > 0) {
        // only first whitespace should be trimmed
        const skip = line[index + 1] === ' ' ? 2 : 1
        field = line.substring(0, index)
        value = line.substring(index + skip)
      } else if (index < 0) {
        // Interpret the entire line as the field name, and use the empty string as the field value
        field = line
        value = ''
      } else {
        // A colon is the first character. This is a comment; ignore it.
        return
      }

      if (!(field in e)) {
        return
      }

      // consecutive 'data' is concatenated with newlines
      if (field === 'data' && e[field] !== null) {
        e['data'] += '\n' + value
      } else {
        e[field] = value
      }
    })

    if (e.id !== null) {
      this.lastEventId = e.id
    }

    const event = new SSECustomEvent(e.event || 'message')
    event.id = e.id
    event.data = e.data || ''
    event.lastEventId = this.lastEventId
    return event
  }
}

if (typeof module !== 'undefined' && module.exports) {
  module.exports = SSEClient
}

export default SSEClient