file/chunk upload

[canvascat]

文件切片上传

前端部分

文件验证

使用文件 hash 值判断，这里使用 js-spark-md5。

对于大文件的 hash 计算，可能导致卡顿（又是单线程的锅 😅），可以参考 hash a file incrementally，当然也可以约定只取文件的一部分来获取 hash。

import SparkMd5 from 'spark-md5';

export const createFileMd5 = (blob: Blob, raw?: boolean) =>
  new Promise(
    (
      resolve: (hash: string) => void,
      reject: (ev: ProgressEvent<FileReader>) => void
    ) => {
      const reader = new FileReader();
      reader.onload = () => {
        const spark = new SparkMd5.ArrayBuffer();
        spark.append(reader.result as ArrayBuffer);
        resolve(spark.end(raw));
      };
      reader.onerror = reject;
      reader.readAsArrayBuffer(blob);
    }
  );

生成文件切片

Blob.slice([start [, end [, contentType]]]) 方法用于创建一个包含源 Blob的指定字节范围内的数据的新 Blob 对象。

export const createChunks = (file: File, chunkSize: number) =>
  [...Array(Math.ceil(file.size / chunkSize))].map((v, i) =>
    file.slice(i++ * chunkSize, Math.min(i * chunkSize, file.size))
  );

文件切片上传

import axios from 'axios';
import { noop, sum } from 'lodash';

const MAX_CHUNK_SIZE = 1024 * 1024 * 3;

const createFormData = (form) =>
  Object.entries(form).reduce(
    (f, [key, val]) => (f.append(key, val as string | Blob), f),
    new FormData()
  );

export const chunkUpload = async (
  file: File,
  url: string,
  onProgress: (
    loaded: number,
    total: number,
    loadeds: Array<number>
  ) => void = noop
) => {
  const md5 = await createFileMd5(file.slice(0, MAX_CHUNK_SIZE));
  const { name, size: total } = file;
  const chunks = createChunks(file, MAX_CHUNK_SIZE);
  const count = chunks.length;
  // 用于保存各切片的上传进度
  const loadeds = Array(count).fill(0);
  await Promise.all(
    chunks.map((chunk, index) => {
      const data = createFormData({ name, total, chunk, index, md5, count });
      return axios
        .post(url, data, {
          onUploadProgress(evt) {
            // loaded 有可能会稍大于 chunk.size ?
            loadeds[index] = Math.min(chunk.size, evt.loaded);
            onProgress(sum(loadeds), total);
          },
        })
        .then(() => {
          loadeds[index] = chunk.size;
          onProgress(sum(loadeds), total);
        });
    })
  );
};

node 部分

文件的接收

这里使用 koa-router

const Router = require('@koa/router');
const { promisify } = require('util');
const fs = require('fs');
const path = require('path');
const { mergeChunks } = require('./util');

const router = new Router();

const uploadPath = path.join(__dirname, 'upload');
const uploadTemp = path.join(__dirname, 'upload/temp');

router.post('/upload/chunks', async (ctx) => {
  const {
    body: { md5, index, count, name },
    files: { chunk },
  } = ctx.request;
  const chunkTempPath = path.resolve(uploadTemp, md5);
  // 创建切片的保存路径
  if (!fs.existsSync(chunkTempPath)) {
    await promisify(fs.mkdir)(chunkTempPath);
  }
  // 将接收的切片移动到创建的路径下，并以 index 命名
  const chunkTemp = path.resolve(chunkTempPath, index);
  if (!fs.existsSync(chunkTemp)) {
    await promisify(fs.rename)(chunk.path, chunkTemp);
  }
  // 读取已保存的切片，如果切片已经全部接收则进行组装
  const cacheChunks = await promisify(fs.readdir)(chunkTempPath);
  if (cacheChunks.length === +count) {
    const dest = path.join(uploadPath, `${md5}${path.extname(name)}`);
    await mergeChunks(chunkTempPath, dest);
  }
  ctx.body = { msg: 'revice chunks success!' };
});

module.exports = router;

切片的组装

const fs = require('fs');
const path = require('path');
const { promisify } = require('util');

const pipStream = (filePath, ws) =>
  new Promise((resolve, reject) =>
    fs
      .createReadStream(filePath)
      .on('end', resolve)
      .on('error', reject)
      .pipe(ws)
  );

/**
 * 合并切片
 * @param {string} chunkDir 切片所在文件夹
 * @param {string} dest 合并后文件位置
 * @param {number} [chunkSize] 切片大小
 * @returns {Promise}
 */
const mergeChunks = async (chunkDir, dest, chunkSize) => {
  if (fs.existsSync(dest)) return;
  const chunks = await promisify(fs.readdir)(chunkDir);
  const chunksPath = chunks
    .sort((a, b) => +a - +b)
    .map((chunk) => path.join(chunkDir, chunk));
  if (typeof chunkSize !== 'number') {
    const stats = await promisify(fs.stat)(chunksPath[0]);
    chunkSize = stats.size;
  }
  await Promise.all(
    chunksPath.map((chunk, i) =>
      pipStream(
        chunk,
        fs.createWriteStream(dest, {
          start: i * chunkSize,
          end: (i + 1) * chunkSize,
        })
      )
    )
  );
  // 删除 chunks temp 文件夹
  await promisify(fs.rmdir)(chunkDir, { recursive: true });
};

module.exports = {
  mergeChunks,
};

TODO

• 文件校验
• 断点续传