如何结合多个管道可写流正确使用Promise

我一眼就看到核心问题所在了——你在Promise.all的回调处理上犯了一个非常常见的错误，导致收尾函数提前执行；另外你的Promise封装也可以简化，毕竟promisePipe本身就返回Promise，没必要手动再包一层。

为什么收尾函数会提前触发？ #

你代码里的这一行是关键错误：

Promise.all(promises).then(final_function(argument));

这里你直接执行了final_function(argument)，而不是把函数作为回调传给then。这就导致final_function会在Promise.all还没等所有加密操作完成时就立刻被调用，完全不符合你“所有操作完成后执行收尾”的预期。正确的做法是把函数引用（或者一个返回函数的表达式）传给then。

修复后的完整代码 #

我帮你重构了代码，解决了这两个问题，同时保留了你原本的业务逻辑：

var promisePipe = require("promisepipe");
var fs = require("fs");
var crypt = require("crypto");

// 补充你代码里缺失的变量（假设这些是你已经定义好的）
const alg = "aes-256-cbc";
const genhashsub = Buffer.from("your-encryption-key", "utf8");
const iv = Buffer.from("your-iv-value", "utf8");

var files = ['/mnt/Storage/test.txt', '/mnt/Storage/test2.txt', '/mnt/Storage/test3.txt']

// 简化Promise创建：直接返回promisePipe的结果，无需手动包裹Promise
var promises = files.map(function(file_enc) {
  var file_out = file_enc + '.locked';
  return promisePipe(
    fs.createReadStream(file_enc),
    crypt.createCipheriv(alg, genhashsub, iv),
    fs.createWriteStream(file_out),
  ).then(function(streams){
    console.log('File written: ' + file_out);
    // 返回当前加密完成的文件路径，会作为Promise的resolve值传给Promise.all
    return file_out;
  }).catch(function(err) {
    const errorMsg = err.message.substring(0, 7) === 'EACCES:' 
      ? `Error (file ${file_out}): Insufficient rights on file or folder`
      : `Error (file ${file_out}): ${err}`;
    console.log(errorMsg);
    // 抛出错误，让Promise.all能捕获到失败的任务
    throw new Error(errorMsg);
  });
});

// 正确传递回调：用箭头函数包裹，确保final_function在所有Promise完成后执行
Promise.all(promises).then(() => final_function(argument))
  .catch(err => {
    // 统一处理所有加密失败的情况
    console.log('One or more encryption tasks failed:', err);
  });

关键修改点说明 #

1. 移除冗余的Promise包裹：promisePipe本身就返回一个Promise，直接返回它的链式调用结果即可，避免了手动resolve/reject可能出现的遗漏或错误。
2. 修正then的回调传递：用() => final_function(argument)箭头函数作为回调，这样final_function只会在Promise.all的所有任务都完成后才会被调用，同时能正确传递argument参数。
3. 规范resolve和reject逻辑：在promisePipe的then里返回file_out，Promise.all的结果数组会包含所有成功加密的文件路径；在catch里抛出错误，让Promise.all能捕获到失败的任务，进入统一的错误处理分支。

额外优化建议（可选） #

如果你的Node.js版本支持，可以用async/await让代码更易读、更简洁：

// 封装成async函数
async function encryptFiles(files) {
  const encryptedFiles = [];
  for (const file_enc of files) {
    const file_out = file_enc + '.locked';
    try {
      await promisePipe(
        fs.createReadStream(file_enc),
        crypt.createCipheriv(alg, genhashsub, iv),
        fs.createWriteStream(file_out),
      );
      console.log('File written: ' + file_out);
      encryptedFiles.push(file_out);
    } catch (err) {
      const errorMsg = err.message.substring(0, 7) === 'EACCES:' 
        ? `Error (file ${file_out}): Insufficient rights on file or folder`
        : `Error (file ${file_out}): ${err}`;
      console.log(errorMsg);
      throw new Error(errorMsg);
    }
  }
  return encryptedFiles;
}

// 调用加密函数并执行收尾逻辑
encryptFiles(files)
  .then(() => final_function(argument))
  .catch(err => console.log('Encryption process failed:', err));

内容的提问来源于stack exchange，提问作者Alex Lias