利用stream为node-fetch实现文件上传下载进度条功能

[54leibo]

• 背景：Electron环境；接口请求node实现；对于上传需要将完整的文件buffer转化为流实现上传进度提示功能
• 依赖
  • bufferhelper
  • progress-stream：生成流的处理进度

  {
      percentage: 9.05, // 下载百分比
      transferred: 949624, // 已传输数据（单位byte）
      length: 10485760, // 数据总长度
      remaining: 9536136, // 剩余未下载
      eta: 42, // 预计剩余时间
      runtime: 3, // 已经消耗时间
      delta: 295396, // 未知
      speed: 949624 // 处理速度
  }
  

  • stream：node 原生模块
• stream to buffer 实现代码

const BufferHelper = require('bufferhelper');
const progress = require('progress-stream');
const { Duplex } = require('stream');

function _streamToBuffer(stream) {
  return new Promise((resolve, reject) => {
    const bufferHelper = new BufferHelper();
    stream.on('error', reject);
    stream.on('data', (chunk) => bufferHelper.concat(chunk));
    stream.on('end', () => resolve(bufferHelper.toBuffer()));
  });
}

const bufferStream = // buffer stream
const bufferSize = // buffer total size
    
const processGenerator = progress({
      length: buffer.byteLength,
      time: 100,
    }, progressInfo => {
      // process progessInfo
    });

_streamToBuffer(bufferStream.pipe(processGenerator))
  .then(() => {
      // success process
  })
  .catch((e) => {
    // error process
  });

• buffer to stream 实现代码

const BufferHelper = require('bufferhelper');
const progress = require('progress-stream');
const { Duplex } = require('stream');

function _bufferToStream(buffer, chunkCount = 100) {
  if (!(buffer instanceof Buffer)) {
    throw new Error(`buffer must be an instance of Buffer, get ${typeof buffer}`)
  }

  if (typeof chunkCount !== 'number') {
    throw new Error(`chunkCount must be a number, get ${typeof chunkCount}`)
  }

  const size = buffer.byteLength;
  let index = 0;
  const chunkSize = Math.ceil(size / chunkCount);

  const stream = new Duplex();
  stream._read = () => {
    if (index < chunkCount) {
      stream.push(buffer.slice(index * chunkSize, (index + 1) * chunkSize));
      // eslint-disable-next-line no-plusplus
      index++
    } else if(index === chunkCount) {
      stream.push(buffer.slice(index * chunkCount, size + 1));
    } else {
      stream.push(null);
    }
  };

  return stream;
}

const buffer = // buffer
const processGenerator = progress({
    length: buffer.byteLength,
    time: 100, /* ms */
  }, progressInfo => {
    options.onProgress(progressInfo);
  });

  try {
    const bufferStream = _bufferToStream(buffer, 100).pipe(processGenerator);
  } catch (e) {
    // error process
  }