79

我写了一个非常简单的函数，它从给定的 URL 下载图像，调整大小并上传到 S3（使用 'gm' 和 'knox'），我不知道我是否正在正确地将流读取到缓冲区. （一切正常，但它是正确的方法吗？）

另外，我想了解有关事件循环的一些信息，我怎么知道函数的一次调用不会泄漏任何内容或将“buf”变量更改为另一个已经运行的调用（或者这种情况是不可能的，因为回调是匿名的功能？）

var http = require('http');
var https = require('https');
var s3 = require('./s3');
var gm = require('gm');

module.exports.processImageUrl = function(imageUrl, filename, callback) {
var client = http;
if (imageUrl.substr(0, 5) == 'https') { client = https; }

client.get(imageUrl, function(res) {
    if (res.statusCode != 200) {
        return callback(new Error('HTTP Response code ' + res.statusCode));
    }

    gm(res)
        .geometry(1024, 768, '>')
        .stream('jpg', function(err, stdout, stderr) {
            if (!err) {
                var buf = new Buffer(0);
                stdout.on('data', function(d) {
                    buf = Buffer.concat([buf, d]);
                });

                stdout.on('end', function() {
                    var headers = {
                        'Content-Length': buf.length
                        , 'Content-Type': 'Image/jpeg'
                        , 'x-amz-acl': 'public-read'
                    };

                    s3.putBuffer(buf, '/img/d/' + filename + '.jpg', headers, function(err, res) {
                        if(err) {
                            return callback(err);
                        } else {
                            return callback(null, res.client._httpMessage.url);
                        }
                    });
                });
            } else {
                callback(err);
            }
        });
    }).on('error', function(err) {
        callback(err);
    });
};
4

9 回答 9

93

总的来说，我没有看到任何会破坏您的代码的内容。

两个建议：

组合Buffer对象的方式不是最理想的，因为它必须在每个“数据”事件上复制所有预先存在的数据。最好将块放在一个数组中，然后concat将它们全部放在最后。

var bufs = [];
stdout.on('data', function(d){ bufs.push(d); });
stdout.on('end', function(){
  var buf = Buffer.concat(bufs);
})

为了提高性能，我会调查您使用的 S3 库是否支持流。理想情况下，您根本不需要创建一个大缓冲区，而只需将stdout流直接传递给 S3 库。

至于你问题的第二部分，那是不可能的。当一个函数被调用时，它被分配了自己的私有上下文，并且其中定义的所有内容都只能从该函数中定义的其他项访问。

更新

将文件转储到文件系统可能意味着每个请求的内存使用量更少，但文件 IO 可能非常慢，因此可能不值得。我想说，在您可以分析和压力测试此功能之前，您不应该优化太多。如果垃圾收集器正在完成它的工作，那么您可能会过度优化。

尽管如此，还是有更好的方法，所以不要使用文件。由于您想要的只是长度，您可以计算它而无需将所有缓冲区附加在一起，因此您根本不需要分配新的缓冲区。

var pause_stream = require('pause-stream');

// Your other code.

var bufs = [];
stdout.on('data', function(d){ bufs.push(d); });
stdout.on('end', function(){
  var contentLength = bufs.reduce(function(sum, buf){
    return sum + buf.length;
  }, 0);

  // Create a stream that will emit your chunks when resumed.
  var stream = pause_stream();
  stream.pause();
  while (bufs.length) stream.write(bufs.shift());
  stream.end();

  var headers = {
      'Content-Length': contentLength,
      // ...
  };

  s3.putStream(stream, ....);
于 2013-01-11T00:05:01.017 回答
25

Javascript 片段

function stream2buffer(stream) {

    return new Promise((resolve, reject) => {
        
        const _buf = [];

        stream.on("data", (chunk) => _buf.push(chunk));
        stream.on("end", () => resolve(Buffer.concat(_buf)));
        stream.on("error", (err) => reject(err));

    });
}

打字稿片段

async function stream2buffer(stream: Stream): Promise<Buffer> {

    return new Promise < Buffer > ((resolve, reject) => {
        
        const _buf = Array < any > ();

        stream.on("data", chunk => _buf.push(chunk));
        stream.on("end", () => resolve(Buffer.concat(_buf)));
        stream.on("error", err => reject(`error converting stream - ${err}`));

    });
}
于 2021-05-27T20:38:48.280 回答
9

如果您从 http(s) URI 中提取，您可以使用node-fetch轻松完成此操作。

从自述文件：

fetch('https://assets-cdn.github.com/images/modules/logos_page/Octocat.png')
    .then(res => res.buffer())
    .then(buffer => console.log)
于 2018-02-05T12:31:21.247 回答
4

您可以将可读流转换为缓冲区，并以这样的异步方式将其集成到您的代码中。

async streamToBuffer (stream) {
    return new Promise((resolve, reject) => {
      const data = [];

      stream.on('data', (chunk) => {
        data.push(chunk);
      });

      stream.on('end', () => {
        resolve(Buffer.concat(data))
      })

      stream.on('error', (err) => {
        reject(err)
      })
   
    })
  }

用法很简单：

 // usage
  const myStream // your stream
  const buffer = await streamToBuffer(myStream) // this is a buffer
于 2021-01-19T14:28:44.373 回答
4

我建议使用 loganfsmyths 方法，使用数组来保存数据。

var bufs = [];
stdout.on('data', function(d){ bufs.push(d); });
stdout.on('end', function(){
  var buf = Buffer.concat(bufs);
}

在我当前的工作示例中，我正在使用 GRIDfs 和 npm 的 Jimp。

   var bucket = new GridFSBucket(getDBReference(), { bucketName: 'images' } );
    var dwnldStream = bucket.openDownloadStream(info[0]._id);// original size
  dwnldStream.on('data', function(chunk) {
       data.push(chunk);
    });
  dwnldStream.on('end', function() {
    var buff =Buffer.concat(data);
    console.log("buffer: ", buff);
       jimp.read(buff)
.then(image => {
         console.log("read the image!");
         IMAGE_SIZES.forEach( (size)=>{
         resize(image,size);
         });
});

我做了一些其他的研究

使用字符串方法但不起作用，可能是因为我正在从图像文件中读取，但数组方法确实有效。

const DISCLAIMER = "DONT DO THIS";
var data = "";
stdout.on('data', function(d){ 
           bufs+=d; 
         });
stdout.on('end', function(){
          var buf = Buffer.from(bufs);
          //// do work with the buffer here

          });

当我执行字符串方法时，我从 npm jimp 收到此错误

buffer:  <Buffer 00 00 00 00 00>
{ Error: Could not find MIME for Buffer <null>

基本上我认为从二进制到字符串的类型强制没有那么好。

于 2019-06-03T19:53:47.700 回答
1

我建议在最后只将缓冲区数组和结果缓冲区连接一次。它很容易手动完成，或者可以使用节点缓冲区

于 2013-01-11T00:03:22.173 回答
1

我只想发布我的解决方案。以前的答案对我的研究很有帮助。我使用 length-stream 来获取流的大小，但这里的问题是回调是在流的末尾附近触发的，所以我也使用 stream-cache 来缓存流并在我知道后将其通过管道传递给 res 对象内容长度。万一出现错误，

var StreamCache = require('stream-cache');
var lengthStream = require('length-stream');

var _streamFile = function(res , stream , cb){
    var cache = new StreamCache();

    var lstream = lengthStream(function(length) {
        res.header("Content-Length", length);
        cache.pipe(res);
    });

    stream.on('error', function(err){
        return cb(err);
    });

    stream.on('end', function(){
        return cb(null , true);
    });

    return stream.pipe(lstream).pipe(cache);
}
于 2014-10-29T09:16:29.383 回答
1

在 ts 中，[].push(bufferPart) 不兼容；

所以：

getBufferFromStream(stream: Part | null): Promise<Buffer> {
    if (!stream) {
        throw 'FILE_STREAM_EMPTY';
    }
    return new Promise(
        (r, j) => {
            let buffer = Buffer.from([]);
            stream.on('data', buf => {
               buffer = Buffer.concat([buffer, buf]);
            });
            stream.on('end', () => r(buffer));
            stream.on('error', j);
        }
    );
}
于 2020-05-22T08:54:47.483 回答
0

您可以在 res.headers中检查“内容长度”标头。它会给你你将收到的内容的长度（它将发送多少字节的数据）

于 2021-04-25T17:35:57.300 回答