28671081942

Committed 03 Jul 2026 03:53PM UTC coverage: 92.881% (-3.7%) from 96.61%

Build # 28671081942

Build Type

Pull #80

github

Committed by

web-flow

Commit Message

Merge 7cfe4e008 into f94b55c6d

Pull Request Pull Request #80: ci: add database version matrix for all active LTS releases

Coverage Stats

40 of 47 branches covered (85.11%)

Branch coverage included in aggregate %.

234 of 248 relevant lines covered (94.35%)

25.52 hits per line

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

96.3

/src/main/common/classes/AbstractBatchEntityWriterStream.ts

import { ObjectWritable, ObjectWritableOptions, WriteCallback, BatchData } from "batchjs";

/**
 * @interface
 * Options for the AbstractBatchEntityWriterStream.
 * @extends ObjectWritableOptions
 */
export interface AbstractBatchEntityWriterStreamOptions extends ObjectWritableOptions {
    /** The maximum number of entities to write at once. */
    batchSize: number;
}

/**
 * @class
 * Class that enable to implement classes to write data in batches of a specified size in different types of data storage.
 * @extends ObjectWritable
 * @template T The type of the data to be written
 */
export abstract class AbstractBatchEntityWriterStream<T> extends ObjectWritable<T> {
    protected buffer: BatchData<T> = [];
    private readonly batchSize: number;

    /**
     
     * @param {AbstractBatchEntityWriterStreamOptions} options - The options for the AbstractBatchEntityWriterStream.
     */
    constructor(options: AbstractBatchEntityWriterStreamOptions) {
        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 {WriteCallback} callback - The callback function to be executed after writing the data.
     */
    _write(chunk: T, encoding: BufferEncoding, callback: WriteCallback): void {
        this.buffer.push(chunk);
        if (this.buffer.length >= this.batchSize) {
            this._flush()
                .then(() => callback())
                .catch((error) => callback(error));
        } else {
            callback();
        }
    }

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

    /**
     * Creates a batch of data from the buffer and flushes it to the storage.
     * 
     * @private
     * @returns {Promise<void>}
     */
    private async _flush(): Promise<void> {
        if (this.buffer.length === 0) return Promise.resolve();
        const batch = [...this.buffer];
        this.buffer = [];
        try {
            await this.batchWrite(batch);
        } catch (error) {
            this.buffer.unshift(...batch);
            return Promise.reject(error as Error);
        }
    }

    /**
     * Writes a batch of data to the storage. This method should be implemented
     * by subclasses to define the specific logic for writing a batch of data.
     *
     * @protected
     * @abstract
     * @param {BatchData<T>} chunk - The batch of data to write to the storage.
     * @returns {Promise<void>} A promise that resolves when the batch is successfully written.
     *                           The promise should be rejected if there is an error during writing.
     */
    protected abstract batchWrite(chunk: BatchData<T>): Promise<void>;

}

1	import { ObjectWritable, ObjectWritableOptions, WriteCallback, BatchData } from "batchjs";	44✔
2
3	/**
4	* @interface
5	* Options for the AbstractBatchEntityWriterStream.
6	* @extends ObjectWritableOptions
7	*/
8	export interface AbstractBatchEntityWriterStreamOptions extends ObjectWritableOptions {
9	/** The maximum number of entities to write at once. */
10	batchSize: number;
11	}
12
13	/**
14	* @class
15	* Class that enable to implement classes to write data in batches of a specified size in different types of data storage.
16	* @extends ObjectWritable
17	* @template T The type of the data to be written
18	*/
19	export abstract class AbstractBatchEntityWriterStream<T> extends ObjectWritable<T> {	44✔
20	protected buffer: BatchData<T> = [];	44✔
21	private readonly batchSize: number;
22
23	/**
24
25	* @param {AbstractBatchEntityWriterStreamOptions} options - The options for the AbstractBatchEntityWriterStream.
26	*/
27	constructor(options: AbstractBatchEntityWriterStreamOptions) {
28	super(options);	44✔
29	this.batchSize = options.batchSize;	44✔
30	}
31
32	/**
33	* A method to write data to the stream, push the chunk to the buffer, and execute the callback.
34	*
35	* @param {T} chunk - The data chunk to write to the stream.
36	* @param {BufferEncoding} encoding - The encoding of the data.
37	* @param {WriteCallback} callback - The callback function to be executed after writing the data.
38	*/
39	_write(chunk: T, encoding: BufferEncoding, callback: WriteCallback): void {
40	this.buffer.push(chunk);	110✔
41	if (this.buffer.length >= this.batchSize) {	110✔
42	this._flush()	22✔
43	.then(() => callback())	22✔
44	.catch((error) => callback(error));	×
45	} else {
46	callback();	88✔
47	}
48	}
49
50	/**
51	* Finalizes the stream by pushing remaining data batches, handling errors,
52	* and executing the final callback.
53	*
54	* @param {WriteCallback} callback - The callback function to be executed after finalizing the stream.
55	* @return {Promise<void>} This function does not return anything.
56	*/
57	_final(callback: WriteCallback): void {
58	this._flush()	44✔
59	.then(() => callback())	22✔
60	.catch((error) => callback(error));	22✔
61	}
62
63	/**
64	* Creates a batch of data from the buffer and flushes it to the storage.
65	*
66	* @private
67	* @returns {Promise<void>}
68	*/
69	private async _flush(): Promise<void> {
70	if (this.buffer.length === 0) return Promise.resolve();	66✔
71	const batch = [...this.buffer];	44✔
72	this.buffer = [];	44✔
73	try {	44✔
74	await this.batchWrite(batch);	44✔
75	} catch (error) {
76	this.buffer.unshift(...batch);	22✔
77	return Promise.reject(error as Error);	22✔
78	}
79	}
80
81	/**
82	* Writes a batch of data to the storage. This method should be implemented
83	* by subclasses to define the specific logic for writing a batch of data.
84	*
85	* @protected
86	* @abstract
87	* @param {BatchData<T>} chunk - The batch of data to write to the storage.
88	* @returns {Promise<void>} A promise that resolves when the batch is successfully written.
89	* The promise should be rejected if there is an error during writing.
90	*/
91	protected abstract batchWrite(chunk: BatchData<T>): Promise<void>;
92
93	}

palcarazm / batchjs-data / 28671081942

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