将流通过管道传输到 s3.upload()答案

【问题标题】：Pipe a stream to s3.upload()将流通过管道传输到 s3.upload()
【发布时间】：2016-09-17 02:11:03
【问题描述】：

我目前正在使用名为 s3-upload-stream 的 node.js 插件将非常大的文件流式传输到 Amazon S3。它使用多部分 API，并且在大多数情况下运行良好。

但是，这个模块已经过时了，我已经不得不对其进行修改（作者也已弃用它）。今天又遇到了亚马逊的问题，很想采纳作者的建议，开始使用官方的aws-sdk来完成我的上传。

但是。

官方 SDK 似乎不支持到s3.upload() 的管道。 s3.upload 的本质是您必须将可读流作为参数传递给 S3 构造函数。

我有大约 120 多个执行各种文件处理的用户代码模块，并且它们不知道其输出的最终目的地。引擎交给他们一个可管道的可写输出流，然后他们通过管道传输给它。我不能给他们一个AWS.S3 对象并要求他们在不向所有模块添加代码的情况下调用upload()。我之所以使用s3-upload-stream，是因为它支持管道。

有没有办法让 aws-sdk s3.upload() 可以通过管道将流传输到？

【问题讨论】：

标签： node.js amazon-web-services amazon-s3 node-streams

【解决方案1】：

创建一个new stream.PassThrough() 和pipe 的输入流给它，然后将传递实例传递给主体。

检查以下示例：

function upload(s3, inputStream) {
    const pass = new PassThrough();

    inputStream.pipe(pass);

    return s3.upload(
        {
            Bucket: 'bucket name',
            Key: 'unique file name',
            Body: pass,
        },
        {
            queueSize: 4, // default concurrency
        },
    ).promise()
        .then((data) => console.log(data))
        .catch((error) => console.error(error));
}

【讨论】：

【解决方案2】：

如果您使用的是 AWS 节点 SDK v3，则有专门的模块用于上传流/blob/缓冲区。

https://www.npmjs.com/package/@aws-sdk/lib-storage

【讨论】：

【解决方案3】：

答案有点晚，希望它可能对其他人有所帮助。您可以同时返回可写流和 Promise，以便在上传完成时获取响应数据。

const AWS = require('aws-sdk');
const stream = require('stream');

const uploadStream = ({ Bucket, Key }) => {
  const s3 = new AWS.S3();
  const pass = new stream.PassThrough();
  return {
    writeStream: pass,
    promise: s3.upload({ Bucket, Key, Body: pass }).promise(),
  };
}

你可以使用如下函数：

const { writeStream, promise } = uploadStream({Bucket: 'yourbucket', Key: 'yourfile.mp4'});
const readStream = fs.createReadStream('/path/to/yourfile.mp4');

const pipeline = readStream.pipe(writeStream);

现在您可以检查承诺：

promise.then(() => {
  console.log('upload completed successfully');
}).catch((err) => {
  console.log('upload failed.', err.message);
});

或者使用异步/等待：

try {
    await promise;
    console.log('upload completed successfully');
} catch (error) {
    console.log('upload failed.', error.message);
}

或者作为stream.pipe()返回stream.Writable，目的地（上面的writeStream变量），允许管道链，我们也可以使用它的事件：

 pipeline.on('close', () => {
   console.log('upload successful');
 });
 pipeline.on('error', (err) => {
   console.log('upload failed', err.message)
 });

【讨论】：

看起来不错，但在我这边我收到了这个错误stackoverflow.com/questions/62330721/…
刚刚回复了您的问题。希望对您有所帮助。
我在异步函数中调用它，所以我使用await Promise。对我有用，谢谢 - 这对我来说是一个巨大而出乎意料的问题。

【解决方案4】：

按照其他答案并使用适用于 Node.js 的最新 AWS 开发工具包，由于 s3 upload() 函数使用 await 语法和 S3 的承诺接受流，因此有一个更清洁、更简单的解决方案：

var model = await s3Client.upload({
    Bucket : bucket,
    Key : key,
    ContentType : yourContentType,
    Body : fs.createReadStream(path-to-file)
}).promise();

【讨论】：

这适用于作者提到的“读取非常大的文件”的特定用例，但如果您在文件上下文之外使用流（例如尝试将 mongo 光标流写入 s3，您仍然需要使用 PassThrough 流 + 管道）

【解决方案5】：

在上面最被接受的答案中需要注意的是：如果您使用管道之类的，则需要在函数中返回传递，

fs.createReadStream(<filePath>).pipe(anyUploadFunction())

function anyUploadFunction () { 
 let pass = new stream.PassThrough();
 return pass // <- Returning this pass is important for the stream to understand where it needs to write to.
}

否则它将默默地进入下一个而不抛出错误，或者会抛出 TypeError: dest.on is not a function 的错误，具体取决于您编写函数的方式

【讨论】：

【解决方案6】：

没有一个答案对我有用，因为我想：

管道进入s3.upload()
将s3.upload() 的结果通过管道传输到另一个流中

接受的答案不支持后者。其他的依赖promise api，在使用流管道时工作起来很麻烦。

这是我对已接受答案的修改。

const s3 = new S3();

function writeToS3({Key, Bucket}) {
  const Body = new stream.PassThrough();

  s3.upload({
    Body,
    Key,
    Bucket: process.env.adpBucket
  })
   .on('httpUploadProgress', progress => {
       console.log('progress', progress);
   })
   .send((err, data) => {
     if (err) {
       Body.destroy(err);
     } else {
       console.log(`File uploaded and available at ${data.Location}`);
       Body.destroy();
     }
  });

  return Body;
}

const pipeline = myReadableStream.pipe(writeToS3({Key, Bucket});

pipeline.on('close', () => {
  // upload finished, do something else
})
pipeline.on('error', () => {
  // upload wasn't successful. Handle it
})

【讨论】：

看起来不错，但在我这边我收到了这个错误 stackoverflow.com/questions/62330721/...

【解决方案7】：

我正在使用 KnexJS，但在使用他们的流 API 时遇到了问题。我终于修复了它，希望以下内容对某人有所帮助。

const knexStream = knex.select('*').from('my_table').stream();
const passThroughStream = new stream.PassThrough();

knexStream.on('data', (chunk) => passThroughStream.write(JSON.stringify(chunk) + '\n'));
knexStream.on('end', () => passThroughStream.end());

const uploadResult = await s3
  .upload({
    Bucket: 'my-bucket',
    Key: 'stream-test.txt',
    Body: passThroughStream
  })
  .promise();

【讨论】：

【解决方案8】：

对于那些抱怨当他们使用 s3 api 上传功能并且零字节文件最终在 s3 上（@Radar155 和 @gabo）的人 - 我也遇到了这个问题。

创建第二个 PassThrough 流并将所有数据从第一个传递到第二个，并将对第二个的引用传递给 s3。您可以通过几种不同的方式做到这一点 - 可能一种肮脏的方式是在第一个流上监听“数据”事件，然后将相同的数据写入第二个流 - 与“结束”事件类似 - 只需调用第二个流上的结束函数。我不知道这是否是 aws api 中的错误、节点版本或其他问题 - 但它对我来说解决了这个问题。

下面是它的外观：

var PassThroughStream = require('stream').PassThrough;
var srcStream = new PassThroughStream();

var rstream = fs.createReadStream('Learning/stocktest.json');
var sameStream = rstream.pipe(srcStream);
// interesting note: (srcStream == sameStream) at this point
var destStream = new PassThroughStream();
// call your s3.upload function here - passing in the destStream as the Body parameter
srcStream.on('data', function (chunk) {
    destStream.write(chunk);
});

srcStream.on('end', function () {
    dataStream.end();
});

【讨论】：

这实际上也对我有用。每当使用分段上传时，S3 上传功能都会默默地“死掉”，但是在使用您的解决方案时它工作得很好（！）。谢谢！ :)
你能提供一些关于为什么需要第二个流的信息吗？

【解决方案9】：

类型脚本解决方案：
本例使用：

import * as AWS from "aws-sdk";
import * as fsExtra from "fs-extra";
import * as zlib from "zlib";
import * as stream from "stream";

和异步功能：

public async saveFile(filePath: string, s3Bucket: AWS.S3, key: string, bucketName: string): Promise<boolean> { 

         const uploadStream = (S3: AWS.S3, Bucket: string, Key: string) => {
            const passT = new stream.PassThrough();
            return {
              writeStream: passT,
              promise: S3.upload({ Bucket, Key, Body: passT }).promise(),
            };
          };
        const { writeStream, promise } = uploadStream(s3Bucket, bucketName, key);
        fsExtra.createReadStream(filePath).pipe(writeStream);     //  NOTE: Addition You can compress to zip by  .pipe(zlib.createGzip()).pipe(writeStream)
        let output = true;
        await promise.catch((reason)=> { output = false; console.log(reason);});
        return output;
}

在某处调用此方法：

let result = await saveFileToS3(testFilePath, someS3Bucket, someKey, someBucketName);

【讨论】：

嗨@dzole vladimirov ....太好了。太感谢了。问候。它帮助我解决了将文件上传到 s3 存储桶的问题

【解决方案10】：

在接受的答案中，函数在上传完成之前结束，因此不正确。下面的代码从可读流中正确管道。

Upload reference

async function uploadReadableStream(stream) {
  const params = {Bucket: bucket, Key: key, Body: stream};
  return s3.upload(params).promise();
}

async function upload() {
  const readable = getSomeReadableStream();
  const results = await uploadReadableStream(readable);
  console.log('upload complete', results);
}

您还可以更进一步，使用ManagedUpload 输出进度信息，如下所示：

const manager = s3.upload(params);
manager.on('httpUploadProgress', (progress) => {
  console.log('progress', progress) // { loaded: 4915, total: 192915, part: 1, key: 'foo.jpg' }
});

ManagedUpload reference

A list of available events

【讨论】：

aws-sdk 现在提供内置于 2.3.0+ 的 Promise，因此您不必再解除它们。 s3.upload(params).promise().then(data => data).catch(error => error);
@DBrown 感谢您的指点！我已经相应地更新了答案。
@tsuz，尝试实施您的解决方案给我一个错误：TypeError: dest.on is not a function，知道为什么吗？
什么是dest.on？你能举个例子吗？ @FireBrand
这表示接受的答案不完整，但它不适用于 s3.upload 的管道，如@Womp 更新的帖子中所示。如果更新此答案以获取其他内容的管道输出，那将非常有帮助！

【解决方案11】：

如果它对我能够成功从客户端流式传输到 s3 的任何人有所帮助：

https://gist.github.com/mattlockyer/532291b6194f6d9ca40cb82564db9d2a

服务器端代码假定req 是一个流对象，在我的例子中，它是从客户端发送的，并在标头中设置了文件信息。

const fileUploadStream = (req, res) => {
  //get "body" args from header
  const { id, fn } = JSON.parse(req.get('body'));
  const Key = id + '/' + fn; //upload to s3 folder "id" with filename === fn
  const params = {
    Key,
    Bucket: bucketName, //set somewhere
    Body: req, //req is a stream
  };
  s3.upload(params, (err, data) => {
    if (err) {
      res.send('Error Uploading Data: ' + JSON.stringify(err) + '\n' + JSON.stringify(err.stack));
    } else {
      res.send(Key);
    }
  });
};

是的，它打破了惯例，但如果你看一下要点，它比我使用 multer、busboy 等发现的任何其他东西都要干净...

+1 表示实用主义，感谢@SalehenRahman 的帮助。

【讨论】：

multer，busboy 处理多部分/表单数据上传。当客户端从 XMLHttpRequest 发送缓冲区作为正文时，req 作为流起作用。
澄清一下，上传是从后端而不是客户端执行的，对吗？
是的，它在后端“管道”流，但它来自前端

【解决方案12】：

使用 node.js stream.PassThrough() 流包装 S3 upload() 函数。

这是一个例子：

inputStream
  .pipe(uploadFromStream(s3));

function uploadFromStream(s3) {
  var pass = new stream.PassThrough();

  var params = {Bucket: BUCKET, Key: KEY, Body: pass};
  s3.upload(params, function(err, data) {
    console.log(err, data);
  });

  return pass;
}

【讨论】：

太好了，这解决了我非常丑陋的 hack =-) 你能解释一下 stream.PassThrough() 的实际作用吗？
执行此操作时您的 PassThrough 流是否关闭？我花了很长时间在 s3.upload 中宣传收盘价以达到我的 PassThrough 流。
上传文件大小为0字节。如果我将相同的数据从源流传输到文件系统，那么一切正常。有什么想法吗？
直通流将获取写入其中的字节并输出它们。这使您可以返回一个可写流，当您写入时 aws-sdk 将从该流中读取。我还会从 s3.upload() 返回响应对象，否则您无法确保上传完成。
这不就像将可读流传递给 Body 但有更多代码吗？ AWS 开发工具包仍将在 PassThrough 流上调用 read()，因此没有真正的管道一直到 S3。唯一的区别是中间有一个额外的流。

【解决方案13】：

如果您知道流的大小，可以使用minio-js 像这样上传流：

  s3Client.putObject('my-bucketname', 'my-objectname.ogg', stream, size, 'audio/ogg', function(e) {
    if (e) {
      return console.log(e)
    }
    console.log("Successfully uploaded the stream")
  })

【讨论】：