我正在破解一个 Node 程序,该程序用于smtp-protocol
捕获 SMTP 电子邮件并处理邮件数据。该库将邮件数据作为流提供,我不知道如何将其转换为字符串。
我目前正在使用 将其写入标准输出stream.pipe(process.stdout, { end: false })
,但正如我所说,我需要字符串中的流数据,一旦流结束,我就可以使用它。
如何将 Node.js 流中的所有数据收集到字符串中?
我正在破解一个 Node 程序,该程序用于smtp-protocol
捕获 SMTP 电子邮件并处理邮件数据。该库将邮件数据作为流提供,我不知道如何将其转换为字符串。
我目前正在使用 将其写入标准输出stream.pipe(process.stdout, { end: false })
,但正如我所说,我需要字符串中的流数据,一旦流结束,我就可以使用它。
如何将 Node.js 流中的所有数据收集到字符串中?
另一种方法是将流转换为承诺(请参阅下面的示例)并使用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)
以上都不适合我。我需要使用 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));
});
希望这比上面的答案更有用:
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 文本产生不可预知的故障(它假定每个字符都适合一个字节),但也许您也不关心这一点。
(这个答案来自几年前,当时它是最好的答案。现在下面有一个更好的答案。我没有跟上node.js,我无法删除这个答案,因为它被标记为“在这个问题上是正确的” “。如果你想向下点击,你想让我做什么?)
关键是使用Readable Streamdata
的andend
事件。收听这些事件:
stream.on('data', (chunk) => { ... });
stream.on('end', () => { ... });
当您收到data
事件时,将新的数据块添加到为收集数据而创建的 Buffer 中。
收到end
事件后,如有必要,将完成的 Buffer 转换为字符串。然后用它做你需要做的事情。
你怎么看待这件事 ?
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");
}
我通常使用这个简单的函数将流转换为字符串:
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
});
还有一个使用 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指出这是可能的。
从 nodejs文档中你应该这样做 - 永远记住一个字符串而不知道编码只是一堆字节:
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);
})
流没有简单的.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);
});
使用流行(每周超过 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
})();
我有更多的运气使用这样的:
let string = '';
readstream
.on('data', (buf) => string += buf.toString())
.on('end', () => console.log(string));
我使用节点v9.11.1
,这是来自回调readstream
的响应。http.get
最干净的解决方案可能是使用“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
})
像流减速器这样的东西呢?
这是一个使用 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'));
})
);
这对我有用,并且基于Node v6.7.0 文档:
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);
})
上面的塞巴斯蒂安 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
列出的所有答案似乎都以流动模式打开可读流,这不是 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
使用您可能已经在项目依赖项中拥有的非常流行的stream-buffers
包,这非常简单:
// 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()));
就我而言,内容类型响应标头是Content-Type: text/plain。所以,我已经从 Buffer 中读取了数据,例如:
let data = [];
stream.on('data', (chunk) => {
console.log(Buffer.from(chunk).toString())
data.push(Buffer.from(chunk).toString())
});