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

78.26

/src/main/common/classes/AbstractBatchEntityReaderStream.ts

import { ObjectReadable, ObjectReadableOptions, BatchData, ExtendableReadableEventMap } from "batchjs";

/**
 * @interface
 * Options for the AbstractBatchEntityReaderStream.
 * @extends ObjectReadableOptions
 */
export interface AbstractBatchEntityReaderStreamOptions extends ObjectReadableOptions {
    /** The maximum number of entities to read at once. */
    batchSize: number;
}

/**
 * @class
 * Class that enable to implement classes to read data in batches of a specified size in different types of data storage.
 * @extends ObjectReadable
 * @template T The type of the data to be read
 */
export abstract class AbstractBatchEntityReaderStream<T> extends ObjectReadable<T, ExtendableReadableEventMap<T,{
    drain: void;
}>> {
    private reading: boolean = false;
    private finished: boolean = false;
    private awaitingDrain: boolean = false;
    protected buffer: BatchData<T> = [];
    private readonly batchSize: number;

    /**
     * @param {AbstractBatchEntityReaderStreamOptions} options - The options for the AbstractBatchEntityReaderStream.
     */
    constructor(options: AbstractBatchEntityReaderStreamOptions) {
        super(options);
        this.batchSize = options.batchSize;
    }

    /**
     * Reads a batch of data from the data storage and pushes it to the consumer stream.
     * If the size parameter is not specified, it reads the number of entities specified in the `batchSize` option.
     * If the size parameter is specified, it reads the minimum of the size and the `batchSize` option.
     * If no data is available, it pushes null to the consumer stream to signal that the end of the stream has been reached.
     * If an error occurs while reading data, it emits an error event to the stream.
     *
     * @param {number} [size] - The size parameter for controlling the read operation.
     */
    _read(size: number): void {
        if (this.reading)
            return;
        this.reading = true;
        this.fetch(Math.min(size, this.batchSize))
            .then((entities) => {
                if (entities.length === 0 && !this.finished) {
                    this.finished=true;
                }
                else {
                    this.buffer.push(...entities);
                }
            })
            .then(()=>this._flush())
            .catch((error) => this.emit("error", error))
            .finally(() => {
                this.reading = false;
            });
    }

    /**
     * Flushes the buffer by pushing its content to the consumer stream. If the consumer stream is not ready to receive data, it waits for the drain event and flushes the buffer again when it is emitted.
     * This function is recursive and will keep flushing the buffer until it is empty.
     *
     * @private
     * @returns {Promise<void>} A promise that resolves when the buffer is flushed.
     */
    private _flush():Promise<void>{
        while (this.buffer.length > 0 && !this.awaitingDrain) {
            const chunk = this.buffer.shift();
            if (!this.push(chunk)) {
                this.awaitingDrain=true;
                const timer = setTimeout(()=>this.emit("drain"), this.drainTimeout);
                this.once("drain", () => {
                    clearTimeout(timer);
                    this.awaitingDrain=false;
                    this._flush();
                });
                return Promise.resolve();
            }
        }
        if(this.buffer.length === 0 && this.finished){
            this.push(null);
        }
        return Promise.resolve();
    };

    /**
     * Abstract method for fetching data from the data storage. This method should be implemented
     * by subclasses to define the specific logic for reading a batch of data.
     * 
     * @protected
     * @abstract
     * @param size {number} - The size parameter for controlling the read operation.
     * @returns {Promise<T[]>} A promise that resolves with an array of entities.
     */
    protected abstract fetch(size: number): Promise<T[]>;

}

1	import { ObjectReadable, ObjectReadableOptions, BatchData, ExtendableReadableEventMap } from "batchjs";	44✔
2
3	/**
4	* @interface
5	* Options for the AbstractBatchEntityReaderStream.
6	* @extends ObjectReadableOptions
7	*/
8	export interface AbstractBatchEntityReaderStreamOptions extends ObjectReadableOptions {
9	/** The maximum number of entities to read at once. */
10	batchSize: number;
11	}
12
13	/**
14	* @class
15	* Class that enable to implement classes to read data in batches of a specified size in different types of data storage.
16	* @extends ObjectReadable
17	* @template T The type of the data to be read
18	*/
19	export abstract class AbstractBatchEntityReaderStream<T> extends ObjectReadable<T, ExtendableReadableEventMap<T,{	44✔
20	drain: void;
21	}>> {
22	private reading: boolean = false;	80✔
23	private finished: boolean = false;	80✔
24	private awaitingDrain: boolean = false;	80✔
25	protected buffer: BatchData<T> = [];	80✔
26	private readonly batchSize: number;
27
28	/**
29	* @param {AbstractBatchEntityReaderStreamOptions} options - The options for the AbstractBatchEntityReaderStream.
30	*/
31	constructor(options: AbstractBatchEntityReaderStreamOptions) {
32	super(options);	80✔
33	this.batchSize = options.batchSize;	80✔
34	}
35
36	/**
37	* Reads a batch of data from the data storage and pushes it to the consumer stream.
38	* If the size parameter is not specified, it reads the number of entities specified in the `batchSize` option.
39	* If the size parameter is specified, it reads the minimum of the size and the `batchSize` option.
40	* If no data is available, it pushes null to the consumer stream to signal that the end of the stream has been reached.
41	* If an error occurs while reading data, it emits an error event to the stream.
42	*
43	* @param {number} [size] - The size parameter for controlling the read operation.
44	*/
45	_read(size: number): void {
46	if (this.reading)	132!
47	return;	×
48	this.reading = true;	132✔
49	this.fetch(Math.min(size, this.batchSize))	132✔
50	.then((entities) => {
51	if (entities.length === 0 && !this.finished) {	132✔
52	this.finished=true;	80✔
53	}
54	else {
55	this.buffer.push(...entities);	52✔
56	}
57	})
58	.then(()=>this._flush())	132✔
59	.catch((error) => this.emit("error", error))	×
60	.finally(() => {
61	this.reading = false;	132✔
62	});
63	}
64
65	/**
66	* Flushes the buffer by pushing its content to the consumer stream. If the consumer stream is not ready to receive data, it waits for the drain event and flushes the buffer again when it is emitted.
67	* This function is recursive and will keep flushing the buffer until it is empty.
68	*
69	* @private
70	* @returns {Promise<void>} A promise that resolves when the buffer is flushed.
71	*/
72	private _flush():Promise<void>{
73	while (this.buffer.length > 0 && !this.awaitingDrain) {	132✔
74	const chunk = this.buffer.shift();	78✔
75	if (!this.push(chunk)) {	78✔
76	this.awaitingDrain=true;	×
77	const timer = setTimeout(()=>this.emit("drain"), this.drainTimeout);	×
78	this.once("drain", () => {	×
79	clearTimeout(timer);	×
80	this.awaitingDrain=false;	×
81	this._flush();	×
82	});
83	return Promise.resolve();	×
84	}
85	}
86	if(this.buffer.length === 0 && this.finished){	132✔
87	this.push(null);	80✔
88	}
89	return Promise.resolve();	132✔
90	};
91
92	/**
93	* Abstract method for fetching data from the data storage. This method should be implemented
94	* by subclasses to define the specific logic for reading a batch of data.
95	*
96	* @protected
97	* @abstract
98	* @param size {number} - The size parameter for controlling the read operation.
99	* @returns {Promise<T[]>} A promise that resolves with an array of entities.
100	*/
101	protected abstract fetch(size: number): Promise<T[]>;
102
103	}

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