node.js - Node.js：ストリームをバッファに読み込む方法は？

Question

指定されたURLから画像をダウンロードし、サイズを変更してS3にアップロードする（「gm」と「knox」を使用）という非常に単純な関数を作成しました。バッファへのストリームの読み取りを正しく行っているかどうかわかりません。。（すべてが機能していますが、それは正しい方法ですか？）

また、イベントループについて何かを理解したいのですが、関数の1つの呼び出しが何もリークしたり、「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);
    });
};

score 93 · Accepted Answer

全体的に、コードを壊すようなものは見当たりません。

2つの提案：

オブジェクトを組み合わせる方法Bufferは、すべての「データ」イベントで既存のすべてのデータをコピーする必要があるため、最適ではありません。チャンクを配列に配置し、concatそれらをすべて最後に配置することをお勧めします。

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

パフォーマンスについては、使用しているS3ライブラリがストリームをサポートしているかどうかを調べます。理想的には、1つの大きなバッファーを作成する必要はなく、代わりにstdoutストリームをS3ライブラリに直接渡すだけです。

あなたの質問の2番目の部分に関しては、それは不可能です。関数が呼び出されると、その関数には独自のプライベートコンテキストが割り当てられ、その中で定義されているすべてのものは、その関数内で定義されている他のアイテムからのみアクセスできます。

アップデート

ファイルをファイルシステムにダンプすると、リクエストごとのメモリ使用量が少なくなる可能性がありますが、ファイル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, ....);

score 25 · Accepted Answer

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));

    });
}

Typescriptスニペット

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}`));

    });
}

score 9 · Accepted Answer

http（s）URIからプルしている場合は、node-fetchを使用してこれを簡単に行うことができます。

READMEから：

fetch('https://assets-cdn.github.com/images/modules/logos_page/Octocat.png')
    .then(res => res.buffer())
    .then(buffer => console.log)

score 4 · Accepted Answer

配列を使用してデータを保持する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

          });

文字列メソッドを実行すると、npmjimpからこのエラーが発生しました

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

基本的に、バイナリから文字列への型の強制はあまりうまく機能しなかったと思います。

score 4 · Accepted Answer

読み取り可能なストリームをバッファーに変換し、このように非同期でコードに統合できます。

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

score 1 · Accepted Answer

バッファの配列を用意し、最後に1回だけ結果のバッファに連結することをお勧めします。手動で行うのは簡単ですが、ノードバッファを使用することもできます

score 1 · Accepted Answer

ソリューションを投稿したいだけです。以前の回答は私の研究に非常に役立ちました。長さストリームを使用してストリームのサイズを取得しますが、ここでの問題は、コールバックがストリームの終わり近くで発生することです。そのため、ストリームキャッシュを使用してストリームをキャッシュし、わかったら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);
}

score 1 · Accepted Answer

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);
        }
    );
}

score 0 · Accepted Answer

「content-length」ヘッダーはres.headersで確認できます。受信するコンテンツの長さ（送信するデータのバイト数）がわかります

node.js - Node.js：ストリームをバッファに読み込む方法は？

9 に答える 9

アップデート

Javascriptスニペット

Typescriptスニペット

Related

Reference