Closed as not planned
Description
Details
So I have a custom transform stream that basically just lets me accumulate data from upstream, and run a async task when it's hit a predefined batch limit. The use case is to stream files and batch insert into a database.
What I understand is that we can use a transform stream either as a true transformer, or just as a writer AKA the end of the stream pipeline.
However, if we want a stream that can behave correctly in either mode, does this just work automatically? If I push to the read side buffer but there is nothing attached on the read side, wouldn't the read side buffer of my transform grow forever? I don't get it. I can't find any conversation about this anywhere online.
Node.js version
Not applicable.
Example code
import { Transform, TransformCallback } from "stream"
export class BatchStream extends Transform {
private readonly batchSize: number
private readonly asyncProcessFunction: (batch: any[]) => Promise<any>
private batch: any[]
constructor(batchSize: number, asyncProcessFunction: (batch: any[]) => Promise<any>) {
super({ objectMode: true })
this.batchSize = batchSize
this.asyncProcessFunction = asyncProcessFunction
this.batch = []
}
_transform(chunk: any, encoding: string, callback: TransformCallback): void {
this.batch.push(chunk)
if (this.batch.length >= this.batchSize) {
void this.processBatch(callback)
} else {
callback()
}
}
_flush(callback: TransformCallback): void {
if (this.batch.length > 0) {
void this.processBatch(callback)
} else {
callback()
}
}
private async processBatch(callback: TransformCallback): Promise<void> {
try {
await this.asyncProcessFunction(this.batch)
this.batch = []
callback()
} catch (err) {
callback(err)
}
}
}Operating system
n/a
Scope
code
Module and version
streams