Node配合WebSocket做多文件下载以及进度回传

// list.jsx
await WebSocketClient.connect((event) => {
 const data = JSON.parse(event.data);
 if (data.event === 'close') {
  this.updateCloseStatusOfProgressBar(list, data);
 } else {
  this.generateProgressBarList(list, data);
 }
});
// src/utils/websocket.client.js
async connect(onmessage, onerror) {
 const socket = this.getSocket();
 return new Promise((resolve) => {
  // ...
 });
}
getSocket() {
 if (!this.socket) {
  this.socket = new WebSocket(
   `ws://localhost:${CONFIG.PORT}&＃63;from=client&id=${clientId}`,
   'echo-protocol',
  );
 }
 return this.socket;
}

// public/Javascript/websocket/websocket.server.js
connectToServer(httpServer) {
 initWsServer(httpServer);
 wsServer.on('request', (request) => {
  // uri: ws://localhost:8888&＃63;from=client&id=xxxx-xxxx-xxxx-xxxx
  logger.info('[ws server] request');
  const cOnnection= request.accept('echo-protocol', request.origin);
  const queryStrings = querystring.parse(request.resource.replace(/(^\/|\&＃63;)/g, ''));
  
  // 每有连接连到websocket服务器，就将当前连接保存到map中
  setConnectionToMap(connection, queryStrings);
  connection.on('message', onMessage);
  connection.on('close', (reasonCode, description) => {
   logger.info(`[ws server] connection closed ${reasonCode} ${description}`);
  });
 });

 wsServer.on('close', (connection, reason, description) => {
  logger.info('[ws server] some connection disconnect.');
  logger.info(reason, description);
 });
}

// list.jsx
http.get(
 'download',
 {
  code,
  filename,
  parent_id: row.id,
  child_id: childId,
  download_url: url,
  client_id: clientId,
 },
);
// routes/api.js
router.get('/download', async (req, res) => {
 const { code, filename } = req.query;
 const url = req.query.download_url;
 const clientId = req.query.client_id;
 const parentId = req.query.parent_id;
 const childId = req.query.child_id;
 const cOnnectionId= `${parentId}-${childId}`;
 const params = {
  code,
  url,
  filename,
  parent_id: parentId,
  child_id: childId,
  client_id: clientId,
 };
 const flag = await AnnieDownloader.download(connectionId, params);
 if (flag) {
  await res.json({ code: 200 });
 } else {
  await res.json({ code: 500, msg: 'download error' });
 }
});
// public/Javascript/annie.js
async download(connectionId, params) {
  //...
 // 当annie下载时，会进行数据监听，这里会用到节流，防止进度回传太快，websocket服务器无法反应
 downloadProcess.stdout.on('data', throttle((chunk) => {
  try {
   if (!chunk) {
    isDownloading = false;
   }
   // 这里主要做的是解析数据，然后发送进度和速度等信息给websocket服务器
   getDownloadInfo(chunk, ws, params);
  } catch (e) {
   downloadSuccess = false;
   WsClient.close(params.client_id, connectionId, 'download error');
   this.stop(connectionId);
   logger.error(`[server annie download] error: ${e}`);
  }
 }, 500, 300));
}

// public/Javascript/websocket/websocket.server.js
function onMessage(message) {
 const data = JSON.parse(message.utf8Data);
 const id = data.client_id;
 if (data.event === 'close') {
  logger.info('[ws server] close event');
  closeConnection(id, data);
 } else {
  getConnectionAndSendProgressToClient(data, id);
 }
}
function getConnectionAndSendProgressToClient(data, clientId) {
 const browserClient = clientsMap.get(clientId);
 // logger.info(`[ws server] send ${JSON.stringify(data)} to client ${clientId}`);
 if (browserClient) {
  const serverClientId = `${data.parent_id}-${data.child_id}`;
  const serverClient = clientsMap.get(serverClientId);
  // 发送从web服务器中传过来的进度、速度给浏览器
  browserClient.send(JSON.stringify(data));
  // 如果进度已经达到了100%
  if (data.progress >= 100) {
   logger.info(`[ws server] file has been download successfully, progress is ${data.progress}`);
   logger.info(`[ws server] server client ${serverClientId} ready to disconnect`);
   // 从clientsMap将当前的这个由web服务器创建的websocket客户端移除
   // 然后关闭当前连接
   // 同时发送下载完成的消息给浏览器
   clientsMap.delete(serverClientId);
   serverClient.send(JSON.stringify({ connectionId: serverClientId, event: 'complete' }));
   serverClient.close('download completed');
  }
 }
}