转换流以将字符串添加到每一行
Transform stream to prepend string to each line
我像这样产生了一个子进程:
const n = cp.spawn('bash');
n.stdout.pipe(process.stdout);
n.stderr.pipe(process.stderr);
我正在寻找一个转换流,这样我就可以在子进程的每一行的开头添加类似“[子进程]”的内容,这样我就知道 stdio 来自子进程而不是父进程.
所以它看起来像:
const getTransformPrepender = function() : Transform {
return ...
}
n.stdout.pipe(getTransformPrepender('[child]')).pipe(process.stdout);
n.stderr.pipe(getTransformPrepender('[child]')).pipe(process.stderr);
有谁知道是否有这样的转换包或如何编写一个?
我有这个:
import * as stream from 'stream';
export default function(pre: string){
let saved = '';
return new stream.Transform({
transform(chunk, encoding, cb) {
cb(null, String(pre) + String(chunk));
},
flush(cb) {
this.push(saved);
cb();
}
});
}
但我担心它在边缘情况下不起作用 - 一个块突发可能不包含整行(对于很长的行)。
看起来问题的答案在这里:https://strongloop.com/strongblog/practical-examples-of-the-new-node-js-streams-api/
但是有了这个附录:
https://twitter.com/the1mills/status/886340747275812865
您可以使用以下方式添加到流中:
https://github.com/ORESoftware/prepend-transform
但它旨在像这样解决手头的问题:
import pt from 'prepend-transform';
import * as cp from 'child_process';
const n = cp.spawn('bash');
n.stdout.pipe(pt('child stdout: ')).pipe(process.stdout);
n.stderr.pipe(pt('child stderr: ')).pipe(process.stderr);
一共需要正确处理三种情况:
- 代表整行的单个块
- 代表多行的单个块
- 仅代表部分行的单个块
这里是解决所有三种情况的算法描述
- 收到一大块数据
- 扫描块中的换行符
- 一旦找到换行符,就将其之前的所有内容(包括换行符)作为一个单行条目发送出去,并进行您需要的任何修改
- 重复直到处理完整个块(没有剩余数据)或直到没有找到额外的换行符(还有一些数据,保存以备后用)
这是一个实际的实现,其中描述了为什么需要它等。
请注意,出于性能原因,我没有将缓冲区转换为经典的 JS 字符串。
const { Transform } = require('stream')
const prefix = Buffer.from('[worker]: ')
const prepender = new Transform({
transform(chunk, encoding, done) {
this._rest = this._rest && this._rest.length
? Buffer.concat([this._rest, chunk])
: chunk
let index
// As long as we keep finding newlines, keep making slices of the buffer and push them to the
// readable side of the transform stream
while ((index = this._rest.indexOf('\n')) !== -1) {
// The `end` parameter is non-inclusive, so increase it to include the newline we found
const line = this._rest.slice(0, ++index)
// `start` is inclusive, but we are already one char ahead of the newline -> all good
this._rest = this._rest.slice(index)
// We have a single line here! Prepend the string we want
this.push(Buffer.concat([prefix, line]))
}
return void done()
},
// Called before the end of the input so we can handle any remaining
// data that we have saved
flush(done) {
// If we have any remaining data in the cache, send it out
if (this._rest && this._rest.length) {
return void done(null, Buffer.concat([prefix, this._rest])
}
},
})
process.stdin.pipe(prepender).pipe(process.stdout)
我像这样产生了一个子进程:
const n = cp.spawn('bash');
n.stdout.pipe(process.stdout);
n.stderr.pipe(process.stderr);
我正在寻找一个转换流,这样我就可以在子进程的每一行的开头添加类似“[子进程]”的内容,这样我就知道 stdio 来自子进程而不是父进程.
所以它看起来像:
const getTransformPrepender = function() : Transform {
return ...
}
n.stdout.pipe(getTransformPrepender('[child]')).pipe(process.stdout);
n.stderr.pipe(getTransformPrepender('[child]')).pipe(process.stderr);
有谁知道是否有这样的转换包或如何编写一个?
我有这个:
import * as stream from 'stream';
export default function(pre: string){
let saved = '';
return new stream.Transform({
transform(chunk, encoding, cb) {
cb(null, String(pre) + String(chunk));
},
flush(cb) {
this.push(saved);
cb();
}
});
}
但我担心它在边缘情况下不起作用 - 一个块突发可能不包含整行(对于很长的行)。
看起来问题的答案在这里:https://strongloop.com/strongblog/practical-examples-of-the-new-node-js-streams-api/
但是有了这个附录: https://twitter.com/the1mills/status/886340747275812865
您可以使用以下方式添加到流中:
https://github.com/ORESoftware/prepend-transform
但它旨在像这样解决手头的问题:
import pt from 'prepend-transform';
import * as cp from 'child_process';
const n = cp.spawn('bash');
n.stdout.pipe(pt('child stdout: ')).pipe(process.stdout);
n.stderr.pipe(pt('child stderr: ')).pipe(process.stderr);
一共需要正确处理三种情况:
- 代表整行的单个块
- 代表多行的单个块
- 仅代表部分行的单个块
这里是解决所有三种情况的算法描述
- 收到一大块数据
- 扫描块中的换行符
- 一旦找到换行符,就将其之前的所有内容(包括换行符)作为一个单行条目发送出去,并进行您需要的任何修改
- 重复直到处理完整个块(没有剩余数据)或直到没有找到额外的换行符(还有一些数据,保存以备后用)
这是一个实际的实现,其中描述了为什么需要它等。
请注意,出于性能原因,我没有将缓冲区转换为经典的 JS 字符串。
const { Transform } = require('stream')
const prefix = Buffer.from('[worker]: ')
const prepender = new Transform({
transform(chunk, encoding, done) {
this._rest = this._rest && this._rest.length
? Buffer.concat([this._rest, chunk])
: chunk
let index
// As long as we keep finding newlines, keep making slices of the buffer and push them to the
// readable side of the transform stream
while ((index = this._rest.indexOf('\n')) !== -1) {
// The `end` parameter is non-inclusive, so increase it to include the newline we found
const line = this._rest.slice(0, ++index)
// `start` is inclusive, but we are already one char ahead of the newline -> all good
this._rest = this._rest.slice(index)
// We have a single line here! Prepend the string we want
this.push(Buffer.concat([prefix, line]))
}
return void done()
},
// Called before the end of the input so we can handle any remaining
// data that we have saved
flush(done) {
// If we have any remaining data in the cache, send it out
if (this._rest && this._rest.length) {
return void done(null, Buffer.concat([prefix, this._rest])
}
},
})
process.stdin.pipe(prepender).pipe(process.stdout)