1 import { Transform, TransformCallback } from 'stream'
3 // Thanks: https://stackoverflow.com/a/45126242
4 class StreamReplacer extends Transform {
5 private pendingChunk: Buffer
7 constructor (private readonly replacer: (line: string) => string) {
11 _transform (chunk: Buffer, _encoding: BufferEncoding, done: TransformCallback) {
13 this.pendingChunk = this.pendingChunk?.length
14 ? Buffer.concat([ this.pendingChunk, chunk ])
19 // As long as we keep finding newlines, keep making slices of the buffer and push them to the
20 // readable side of the transform stream
21 while ((index = this.pendingChunk.indexOf('\n')) !== -1) {
22 // The `end` parameter is non-inclusive, so increase it to include the newline we found
23 const line = this.pendingChunk.slice(0, ++index)
25 // `start` is inclusive, but we are already one char ahead of the newline -> all good
26 this.pendingChunk = this.pendingChunk.slice(index)
28 // We have a single line here! Prepend the string we want
29 this.push(this.doReplace(line))
38 _flush (done: TransformCallback) {
39 // If we have any remaining data in the cache, send it out
40 if (!this.pendingChunk?.length) return done()
43 return done(null, this.doReplace(this.pendingChunk))
49 private doReplace (buffer: Buffer) {
50 const line = this.replacer(buffer.toString('utf8'))
52 return Buffer.from(line, 'utf8')