17580344153

Committed 09 Sep 2025 10:55AM UTC coverage: 98.387% (-1.4%) from 99.825%

Build # 17580344153

Build Type

Pull #58

github

Committed by

web-flow

Commit Message

Merge 5a02e8636 into a2981500d

Pull Request Pull Request #58: fix: prevent write callback call before flushing

Coverage Stats

89 of 90 branches covered (98.89%)

Branch coverage included in aggregate %.

31 of 39 new or added lines in 8 files covered. (79.49%)

460 of 468 relevant lines covered (98.29%)

75.42 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.59

/src/streams/classes/BufferStream.ts

import { TransformCallback } from "stream";
import { ObjectDuplex, ObjectDuplexOptions } from "../interfaces/_index";

/**
 * @interface
 * Options for the BufferStream.
 * @extends ObjectDuplexOptions
 */
export interface BufferStreamOptions extends ObjectDuplexOptions {
    batchSize: number;
}

/**
 * @class
 * Class that allows you  stream data in batches of a specified size.
 * @extends ObjectDuplex
 * @template T
 * @example
 * ```typescript
 * const stream:BufferStream<string> = new BufferStream({
 *     objectMode: true,
 *     batchSize: 2,
 * });
 * 
 * stream.write("data1");
 * stream.write("data2");
 * stream.write("data3");
 * stream.end();
 * 
 * stream.on("data", (chunk: Array<string>) => {
 *     console.log(``Pushed chunk: ${JSON.stringify(chunk)}```);
 * });
 * ```
 * ```shell
 * >> Pushed chunk: ["data1", "data2"]
 * >> Pushed chunk: ["data3"]
 * ```
 */
export class BufferStream<T> extends ObjectDuplex<T,T[]> {
    protected buffer: T[] = [];
    private readonly batchSize: number;

    /**
     * @constructor
     * @param {BufferStreamOptions} options - The options for the BufferStream.
     * @param [options.batchSize] {number} - The maximum number of elements in a batch.
     */
    constructor(options: BufferStreamOptions) {
        super(options);
        this.batchSize = options.batchSize;
    }

    /**
     * A method to write data to the stream, push the chunk to the buffer, and execute the callback.
     *
     * @param {T} chunk - The data chunk to write to the stream.
     * @param {BufferEncoding} encoding - The encoding of the data.
     * @param {TransformCallback} callback - The callback function to be executed after writing the data.
     * @return {void} This function does not return anything.
     */
    _write(chunk: T, encoding: BufferEncoding, callback: TransformCallback): void {
        this.buffer.push(chunk);
        this._flush(false)
            .then(()=>callback())
            .catch((e)=>callback(e));
    }

    /**
     * Finalizes the stream by pushing remaining data batches, handling errors,
     * and executing the final callback.
     *
     * @param {TransformCallback} callback - The callback function to be executed after finalizing the stream.
     * @return {void} This function does not return anything.
     */
    _final(callback: TransformCallback): void {
        this._flush(true)
            .then(()=>{
                this.push(null);
                callback();
            }).catch(e=>callback(e));
    }

    /**
     * Pushes the ready chunks to the consumer stream since the buffer is empty or the size limit is reached.
     *
     * @param {number} size - The size parameter for controlling the read operation.
     * @return {void} This function does not return anything.
     */
    _read(): void {
        this._flush(false);
    }

    /**
     * Pushes the next batch of elements from the buffer to the stream, handling backpressure.
     * @protected
     * @return {Promise<void>} This function does not return anything.
     */
    protected async _flush(final:boolean): Promise<void> {
        while (this.buffer.length >= this.batchSize || (final && this.buffer.length > 0)) {
            const batch = this.buffer.splice(0, this.batchSize);
            if(!this.push(batch)){
                await new Promise<void>((resolve) => {  
                    this.once("drain", () => resolve());
                });
            };
        }
        return Promise.resolve();
    }
}

1	import { TransformCallback } from "stream";
2	import { ObjectDuplex, ObjectDuplexOptions } from "../interfaces/_index";	120✔
3
4	/**
5	* @interface
6	* Options for the BufferStream.
7	* @extends ObjectDuplexOptions
8	*/
9	export interface BufferStreamOptions extends ObjectDuplexOptions {
10	batchSize: number;
11	}
12
13	/**
14	* @class
15	* Class that allows you stream data in batches of a specified size.
16	* @extends ObjectDuplex
17	* @template T
18	* @example
19	* ```typescript
20	* const stream:BufferStream<string> = new BufferStream({
21	* objectMode: true,
22	* batchSize: 2,
23	* });
24	*
25	* stream.write("data1");
26	* stream.write("data2");
27	* stream.write("data3");
28	* stream.end();
29	*
30	* stream.on("data", (chunk: Array<string>) => {
31	* console.log(``Pushed chunk: ${JSON.stringify(chunk)}```);
32	* });
33	* ```
34	* ```shell
35	* >> Pushed chunk: ["data1", "data2"]
36	* >> Pushed chunk: ["data3"]
37	* ```
38	*/
39	export class BufferStream<T> extends ObjectDuplex<T,T[]> {	120✔
40	protected buffer: T[] = [];	18✔
41	private readonly batchSize: number;
42
43	/**
44	* @constructor
45	* @param {BufferStreamOptions} options - The options for the BufferStream.
46	* @param [options.batchSize] {number} - The maximum number of elements in a batch.
47	*/
48	constructor(options: BufferStreamOptions) {
49	super(options);	18✔
50	this.batchSize = options.batchSize;	18✔
51	}
52
53	/**
54	* A method to write data to the stream, push the chunk to the buffer, and execute the callback.
55	*
56	* @param {T} chunk - The data chunk to write to the stream.
57	* @param {BufferEncoding} encoding - The encoding of the data.
58	* @param {TransformCallback} callback - The callback function to be executed after writing the data.
59	* @return {void} This function does not return anything.
60	*/
61	_write(chunk: T, encoding: BufferEncoding, callback: TransformCallback): void {
62	this.buffer.push(chunk);	150✔
63	this._flush(false)	150✔
64	.then(()=>callback())	150✔
NEW 65	.catch((e)=>callback(e));	×
66	}
67
68	/**
69	* Finalizes the stream by pushing remaining data batches, handling errors,
70	* and executing the final callback.
71	*
72	* @param {TransformCallback} callback - The callback function to be executed after finalizing the stream.
73	* @return {void} This function does not return anything.
74	*/
75	_final(callback: TransformCallback): void {
76	this._flush(true)	18✔
77	.then(()=>{
78	this.push(null);	18✔
79	callback();	18✔
NEW 80	}).catch(e=>callback(e));	×
81	}
82
83	/**
84	* Pushes the ready chunks to the consumer stream since the buffer is empty or the size limit is reached.
85	*
86	* @param {number} size - The size parameter for controlling the read operation.
87	* @return {void} This function does not return anything.
88	*/
89	_read(): void {
90	this._flush(false);	18✔
91	}
92
93	/**
94	* Pushes the next batch of elements from the buffer to the stream, handling backpressure.
95	* @protected
96	* @return {Promise<void>} This function does not return anything.
97	*/
98	protected async _flush(final:boolean): Promise<void> {
99	while (this.buffer.length >= this.batchSize \|\| (final && this.buffer.length > 0)) {	186✔
100	const batch = this.buffer.splice(0, this.batchSize);	78✔
101	if(!this.push(batch)){	78✔
102	await new Promise<void>((resolve) => {	18✔
103	this.once("drain", () => resolve());	18✔
104	});
105	};
106	}
107	return Promise.resolve();	186✔
108	}
109	}

palcarazm / batchjs / 17580344153

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous