如何将 Node.js 流的内容读入字符串变量？

Question

我正在破解一个 Node 程序，该程序使用smtp-protocol 来捕获 SMTP 电子邮件并处理邮件数据。该库将邮件数据作为流提供，我不知道如何将其转换为字符串。

我目前正在使用stream.pipe(process.stdout, { end: false }) 将其写入标准输出，但正如我所说，我需要字符串中的流数据，一旦流结束我就可以使用它。

如何将 Node.js 流中的所有数据收集到字符串中？

Answer 1

您对此有何看法？

async function streamToString(stream) {
    // lets have a ReadableStream as a stream variable
    const chunks = [];

    for await (const chunk of stream) {
        chunks.push(Buffer.from(chunk));
    }

    return Buffer.concat(chunks).toString("utf-8");
}

Answer 2

另一种方法是将流转换为承诺（请参阅下面的示例）并使用then（或await）将解析后的值分配给变量。

function streamToString (stream) {
  const chunks = [];
  return new Promise((resolve, reject) => {
    stream.on('data', (chunk) => chunks.push(Buffer.from(chunk)));
    stream.on('error', (err) => reject(err));
    stream.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')));
  })
}

const result = await streamToString(stream)

Answer 3

（这个答案是几年前的，当时它是最好的答案。现在下面有一个更好的答案。我没有跟上node.js，我无法删除这个答案，因为它被标记为“正确”关于这个问题”。如果你正在考虑向下点击，你想让我做什么？）

关键是使用Readable Stream 的data 和end 事件。收听这些事件：

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

当您收到data 事件时，将新的数据块添加到为收集数据而创建的 Buffer 中。

当您收到end 事件时，如有必要，将完成的 Buffer 转换为字符串。然后用它做你需要做的事情。

Answer 4

还有一个使用 Promise 的字符串：

function getStream(stream) {
  return new Promise(resolve => {
    const chunks = [];

    # Buffer.from is required if chunk is a String, see comments
    stream.on("data", chunk => chunks.push(Buffer.from(chunk)));
    stream.on("end", () => resolve(Buffer.concat(chunks).toString()));
  });
}

用法：

const stream = fs.createReadStream(__filename);
getStream(stream).then(r=>console.log(r));

如果需要，请删除 .toString() 以与二进制数据一起使用。

更新：@AndreiLED 正确指出字符串存在问题。我无法使用我拥有的节点版本获得返回字符串的流，但api 指出这是可能的。

Answer 5

列出的所有答案似乎都以流动模式打开可读流，这不是 NodeJS 中的默认设置，并且可能存在限制，因为它缺乏 NodeJS 在暂停可读流模式下提供的背压支持。 这是一个使用 Just Buffers、Native Stream 和 Native Stream Transforms 并支持对象模式的实现

import {Transform} from 'stream';

let buffer =null;    

function objectifyStream() {
    return new Transform({
        objectMode: true,
        transform: function(chunk, encoding, next) {

            if (!buffer) {
                buffer = Buffer.from([...chunk]);
            } else {
                buffer = Buffer.from([...buffer, ...chunk]);
            }
            next(null, buffer);
        }
    });
}

process.stdin.pipe(objectifyStream()).process.stdout

Answer 6

在我的例子中，内容类型响应标头是 Content-Type: text/plain。所以，我已经从 Buffer 中读取了数据，例如：

let data = [];
stream.on('data', (chunk) => {
 console.log(Buffer.from(chunk).toString())
 data.push(Buffer.from(chunk).toString())
});

Answer 7

希望这比上面的答案更有用：

var string = '';
stream.on('data',function(data){
  string += data.toString();
  console.log('stream data ' + part);
});

stream.on('end',function(){
  console.log('final output ' + string);
});

请注意，字符串连接并不是收集字符串部分最有效的方法，但它是为了简单起见（也许您的代码并不关心效率）。

此外，此代码可能会为非 ASCII 文本产生不可预知的故障（它假定每个字符都适合一个字节），但也许您也不关心这一点。

Answer 8

setEncoding('utf8');

上面的塞巴斯蒂安 J 做得好。

我有几行测试代码的“缓冲区问题”，并添加了编码信息并解决了它，见下文。

证明问题

软件

// process.stdin.setEncoding('utf8');
process.stdin.on('data', (data) => {
    console.log(typeof(data), data);
});

输入

hello world

输出

object <Buffer 68 65 6c 6c 6f 20 77 6f 72 6c 64 0d 0a>

演示解决方案

软件

process.stdin.setEncoding('utf8'); // <- Activate!
process.stdin.on('data', (data) => {
    console.log(typeof(data), data);
});

输入

hello world

输出

string hello world

Answer 9

使用流行（每周超过 500 万次下载）和轻量级 get-stream 库的简单方法：

https://www.npmjs.com/package/get-stream

const fs = require('fs');
const getStream = require('get-stream');

(async () => {
    const stream = fs.createReadStream('unicorn.txt');
    console.log(await getStream(stream)); //output is string
})();

Answer 10

使用您可能已经在项目依赖项中拥有的quite popular stream-buffers package，这非常简单：

// imports
const { WritableStreamBuffer } = require('stream-buffers');
const { promisify } = require('util');
const { createReadStream } = require('fs');
const pipeline = promisify(require('stream').pipeline);

// sample stream
let stream = createReadStream('/etc/hosts');

// pipeline the stream into a buffer, and print the contents when done
let buf = new WritableStreamBuffer();
pipeline(stream, buf).then(() => console.log(buf.getContents().toString()));

Answer 11

我有更多的运气使用这样的：

let string = '';
readstream
    .on('data', (buf) => string += buf.toString())
    .on('end', () => console.log(string));

我使用节点v9.11.1，readstream 是来自http.get 回调的响应。

Answer 12

最干净的解决方案可能是使用“string-stream”包，它将流转换为带有承诺的字符串。

const streamString = require('stream-string')

streamString(myStream).then(string_variable => {
    // myStream was converted to a string, and that string is stored in string_variable
    console.log(string_variable)

}).catch(err => {
     // myStream emitted an error event (err), so the promise from stream-string was rejected
    throw err
})

Answer 13

流缩减器之类的东西怎么样？

这是一个使用 ES6 类的示例，如何使用一个。

var stream = require('stream')

class StreamReducer extends stream.Writable {
  constructor(chunkReducer, initialvalue, cb) {
    super();
    this.reducer = chunkReducer;
    this.accumulator = initialvalue;
    this.cb = cb;
  }
  _write(chunk, enc, next) {
    this.accumulator = this.reducer(this.accumulator, chunk);
    next();
  }
  end() {
    this.cb(null, this.accumulator)
  }
}

// just a test stream
class EmitterStream extends stream.Readable {
  constructor(chunks) {
    super();
    this.chunks = chunks;
  }
  _read() {
    this.chunks.forEach(function (chunk) { 
        this.push(chunk);
    }.bind(this));
    this.push(null);
  }
}

// just transform the strings into buffer as we would get from fs stream or http request stream
(new EmitterStream(
  ["hello ", "world !"]
  .map(function(str) {
     return Buffer.from(str, 'utf8');
  })
)).pipe(new StreamReducer(
  function (acc, v) {
    acc.push(v);
    return acc;
  },
  [],
  function(err, chunks) {
    console.log(Buffer.concat(chunks).toString('utf8'));
  })
);

Answer 14

这对我有用，基于Node v6.7.0 docs：

let output = '';
stream.on('readable', function() {
    let read = stream.read();
    if (read !== null) {
        // New stream data is available
        output += read.toString();
    } else {
        // Stream is now finished when read is null.
        // You can callback here e.g.:
        callback(null, output);
    }
});

stream.on('error', function(err) {
  callback(err, null);
})

Answer 15

我通常使用这个简单的函数将流转换为字符串：

function streamToString(stream, cb) {
  const chunks = [];
  stream.on('data', (chunk) => {
    chunks.push(chunk.toString());
  });
  stream.on('end', () => {
    cb(chunks.join(''));
  });
}

使用示例：

let stream = fs.createReadStream('./myFile.foo');
streamToString(stream, (data) => {
  console.log(data);  // data is now my string variable
});

Answer 16

Streams 没有简单的 .toString() 函数（我理解），也没有类似 .toStringAsync(cb) 的函数（我不理解）。

所以我创建了自己的辅助函数：

var streamToString = function(stream, callback) {
  var str = '';
  stream.on('data', function(chunk) {
    str += chunk;
  });
  stream.on('end', function() {
    callback(str);
  });
}

// how to use:
streamToString(myStream, function(myStr) {
  console.log(myStr);
});

Answer 17

以上都不适合我。我需要使用 Buffer 对象：

  const chunks = [];

  readStream.on("data", function (chunk) {
    chunks.push(chunk);
  });

  // Send the buffer or you can put it into a var
  readStream.on("end", function () {
    res.send(Buffer.concat(chunks));
  });

Answer 18

从 nodejs documentation 你应该这样做 - 永远记住一个字符串而不知道编码只是一堆字节：

var readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', function(chunk) {
  assert.equal(typeof chunk, 'string');
  console.log('got %d characters of string data', chunk.length);
})