网络优化之 WebSocket

/**
 * Dispatch a request "version": "0.19.2"
 * @param {Object} config The config specific for this request (merged with this.defaults)
 */
Axios.prototype.request = function request(config) {
  // Allow for axios('example/url'[, config]) a la fetch API
  if (typeof config === "string") {
    config = arguments[1] || {};
    config.url = arguments[0];
  } else {
    config = config || {};
  }

  config = mergeConfig(this.defaults, config);

  // Set config.method
  if (config.method) {
    config.method = config.method.toLowerCase();
  } else if (this.defaults.method) {
    config.method = this.defaults.method.toLowerCase();
  } else {
    config.method = "get";
  }

  // Hook up interceptors middleware
  var chain = [dispatchRequest, undefined];
  var promise = Promise.resolve(config);

  this.interceptors.request.forEach(function unshiftRequestInterceptors(
    interceptor
  ) {
    chain.unshift(interceptor.fulfilled, interceptor.rejected);
  });

  this.interceptors.response.forEach(function pushResponseInterceptors(
    interceptor
  ) {
    chain.push(interceptor.fulfilled, interceptor.rejected);
  });

  while (chain.length) {
    promise = promise.then(chain.shift(), chain.shift());
  }

  return promise;
};

async function send(requestParams, successCallback, errorCallback, finalCallback) {
  if (this.isWebSocketOpen) {
    const { seqId, request, params } = requestParams;

    let modifiedConfig = getRequestConfig(apiConfig, params);
    for (let index = 0; index < this.requestInterceptors.length; index++) {
      modifiedConfig = await this.requestInterceptors[index](modifiedConfig);
    }
    const requestFrame = getRequestFrame(seqId, modifiedConfig);

    try {
      const buffer = getRequestBuffer(requestFrame);
      // 回调和超时控制
      this.requestInfoMap.set(seqId, {
        config: modifiedConfig,
        callbacks: [successCallback, errorCallback, finalCallback],
        timer: needClearSetTimeout(() => {
          errorCallback();
          finalCallback();
          const requestInfo = this.requestInfoMap.get(seqId);
          if (requestInfo) {
            requestInfo.isTimedOut = true;
          }
        }, REQUEST_TIMEOUT),
        isTimedOut: false,
      });

      this.socket.send(
        buffer.buffer.slice(buffer.byteOffset, buffer.byteOffset + buffer.length)
      );
    } catch (e) {}
  } else {
    this.reconnect();
  }
}

export function dispatchResponse(
  responseContent: ResponseContent,
  config: RequestConfig,
  devInfo: Record<string, any>
) {
  try {
    const res = JSON.parse(responseContent.response);
    const status = responseContent.statusCode;
    if (status >= 200 && status < 300) {
      if (devInfo.wsAckTime && res.time_consuming) {
        devInfo.wsNetworkTime = devInfo.wsAckTime - res.time_consuming;
      }

      return Promise.resolve({
        status,
        data: res,
        config,
        devInfo,
      });
    } else {
      return Promise.reject({
        status,
        config,
        response: {
          status,
          data: res,
        },
      });
    }
  } catch (error) {
    return Promise.reject(error);
  }
}

handleResponse(
    responseContent: ResponseContent,
    config: RequestConfig,
    devInfo: Record<string, any>
) {
  const responseInterceptorsChain = [] as any[];
  this.responseInterceptors.forEach((interceptor) => {
    responseInterceptorsChain.push(
      interceptor.fulfilled,
      interceptor.rejected
    );
  });

  const len = responseInterceptorsChain.length;
  let promise = dispatchResponse(responseContent, config, devInfo);
  let i = 0;

  while (i < len) {
    promise = promise.then(
      responseInterceptorsChain[i++],
      responseInterceptorsChain[i++]
    );
  }

  return promise;
}

this.socket.onmessage = (event) => {
  const rawData = event.data;

  let responseObject;
  try {
    responseObject = getResponseObject(rawData, true);
  } catch (error) {
    return;
  }
  
  // 处理 pong 则 return

  const seqId = responseObject.seqId;
  if (this.checkResponseTimeout(seqId)) {
    return;
  }
  const requestInfo = this.requestInfoMap.get(seqId);
  store.sessionStore.seqIdToAckMap.delete(seqId);

  if (requestInfo?.callbacks?.length) {
    const [successCallback, errorCallback, finalCallback] =
      requestInfo.callbacks;

    this.handleResponse(
      responseObject.content?.wsProxyHttpResponse || {},
      requestInfo?.config
    )
      .then(successCallback, errorCallback)
      .finally(finalCallback);

    this.requestInfoMap.delete(seqId);
  }
};

connect(openCallback?: Function) {
  if (this.numberOfConnectLeft <= 0) {
    return;
  }

  this.socket = new WebSocket(this.url);
  // An ArrayBuffer would have to be fully read before it 
  // could be used, and all of the data would have to be
  // in memory at the same time. In contrast, a Blob can
  // be read as a stream, avoiding having to have all 
  // the data in memory before you can do anything with
  // it (or at all).
  this.socket.binaryType = 'arraybuffer';

  this.closeTimer = needClearSetTimeout(() => {
    this.close();
    this.isEnabled = false;
  }, REQUEST_TIMEOUT) as unknown as number;

  this.socket.onopen = () => {
    if (!this.isEnabled) {
      return;
    }
    if (this.closeTimer) {
      clearTimeout(this.closeTimer);
    }
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
    }

    openCallback?.();
    this.ping();
  };

  this.socket.onmessage = (event) => {
    // handle response
  };

  this.socket.onclose = (e) => {
    this.reconnect();
  };

  this.socket.onerror = (e) => {
    this.reconnect();
  };

  this.numberOfConnectLeft--;
}

async reconnect() {
  const { isConnected } = await fetchNetworkInfo();
  if (
    !this.isEnabled ||
    !isConnected ||
    this.isReconnectScheduled ||
    this.isWebSocketConnecting
  ) {
    return;
  }

  this.isReconnectScheduled = true;

  this.reconnectTimer = SetTimeout(() => {
    this.connect(tryToPullLostWSMessage);
    this.isReconnectScheduled = false;
    this.reconnectTimeout += RECONNECT_TIMEOUT_INCREASE_UNIT;
  }, this.reconnectTimeout);
}