14983613232

Committed 12 May 2025 10:09PM UTC coverage: 90.019% (+2.2%) from 87.809%

Build # 14983613232

Build Type

push

github

Committed by

palcarazm

Commit Message

1.1.3

Run Details

87 of 111 branches covered (78.38%)

Branch coverage included in aggregate %.

391 of 420 relevant lines covered (93.1%)

43.47 hits per line

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

96.0

/src/streams/classes/GroupByStream.ts

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

/**
 * @interface
 * Options for the GroupByStream.
 * @extends ObjectDuplexOptions
 * @template T
 */
export interface GroupByStreamOptions<T> extends ObjectDuplexOptions {
    groupBy: (chunk: T) => string;
}

/**
 * @class
 * Class that allows you to group data in a stream.
 * @extends ObjectDuplex
 * @template T
 * @example
 * ```typescript
 * const stream:GroupByStream<string> = new GroupByStream({
 *     objectMode: true,
 *     groupBy: (chunk: string) => chunk.split("").at(0) ?? "",
 * });
 * 
 * stream.write("DATA1"); //group : D
 * stream.write("DATA2"); //group : D
 * stream.write("data3"); //group : d
 * stream.end();
 * 
 * stream.on("data", (chunk: Array<string>) => {
 *     console.log(``Pushed chunk: ${chunk}```);
 * });
 * ```
 * ```shell
 * >> Pushed chunk: ["DATA1", "DATA2"]
 * >> Pushed chunk: ["data3"]
 * ```
 */
export class GroupByStream<T> extends ObjectDuplex {
    protected buffer: Map<string,Array<T>> = new Map();
    private readonly groupBy: (chunk: T) => string;

    /**
     * @constructor
     * @param {GroupByStreamOptions<T>} options - The options for the GroupBy.
     * @param [options.groupBy] {Function} - The function used to get the grouping key from the chunk.
     */
    constructor(options: GroupByStreamOptions<T>) {
        super(options);
        this.groupBy = options.groupBy;
    }

    /**
     * Writes a chunk of data to the stream, groups it according to a specified function,
     * and executes 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}
     */
    _write(chunk: T, encoding: BufferEncoding, callback: TransformCallback): void {
        this._groupBy(chunk);
        callback();
    }

 
    /**
     * Finalize the stream by draining the buffer and pushing any remaining chunks to the stream.
     *
     * @param {TransformCallback} callback - The callback to be called when the stream is finalized.
     * @return {void}
     */
    _final(callback: TransformCallback): void {
        while (this.buffer.size > 0) {
            const groupEntry = this.buffer.entries().next().value;
            if (groupEntry) {
                const [groupKey, group] = groupEntry;
                this.push(group);
                this.buffer.delete(groupKey);
            }
        }
        this.push(null);
        callback();
    }

    /**
     * Reading is not supported since writer finishes first.
     *
     * @return {void}
     */
    _read(): void {
        return;
    }

    /**
     * Groups a chunk of data based on the provided groupBy function and stores it in the buffer.
     *
     * @param {T} chunk - The data chunk to be grouped.
     * @return {void} This function does not return anything.
     */
    private _groupBy(chunk: T): void {
        const groupKey = this.groupBy(chunk);
        const group = this.buffer.get(groupKey) ?? [];
        group.push(chunk);
        this.buffer.set(groupKey, group);
    }
}

1	import { TransformCallback } from "stream";
2	import { ObjectDuplex, ObjectDuplexOptions } from "../interfaces/_index";	108✔
3
4	/**
5	* @interface
6	* Options for the GroupByStream.
7	* @extends ObjectDuplexOptions
8	* @template T
9	*/
10	export interface GroupByStreamOptions<T> extends ObjectDuplexOptions {
11	groupBy: (chunk: T) => string;
12	}
13
14	/**
15	* @class
16	* Class that allows you to group data in a stream.
17	* @extends ObjectDuplex
18	* @template T
19	* @example
20	* ```typescript
21	* const stream:GroupByStream<string> = new GroupByStream({
22	* objectMode: true,
23	* groupBy: (chunk: string) => chunk.split("").at(0) ?? "",
24	* });
25	*
26	* stream.write("DATA1"); //group : D
27	* stream.write("DATA2"); //group : D
28	* stream.write("data3"); //group : d
29	* stream.end();
30	*
31	* stream.on("data", (chunk: Array<string>) => {
32	* console.log(``Pushed chunk: ${chunk}```);
33	* });
34	* ```
35	* ```shell
36	* >> Pushed chunk: ["DATA1", "DATA2"]
37	* >> Pushed chunk: ["data3"]
38	* ```
39	*/
40	export class GroupByStream<T> extends ObjectDuplex {	108✔
41	protected buffer: Map<string,Array<T>> = new Map();	12✔
42	private readonly groupBy: (chunk: T) => string;
43
44	/**
45	* @constructor
46	* @param {GroupByStreamOptions<T>} options - The options for the GroupBy.
47	* @param [options.groupBy] {Function} - The function used to get the grouping key from the chunk.
48	*/
49	constructor(options: GroupByStreamOptions<T>) {
50	super(options);	12✔
51	this.groupBy = options.groupBy;	12✔
52	}
53
54	/**
55	* Writes a chunk of data to the stream, groups it according to a specified function,
56	* and executes the callback.
57	*
58	* @param {T} chunk - The data chunk to write to the stream.
59	* @param {BufferEncoding} encoding - The encoding of the data.
60	* @param {TransformCallback} callback - The callback function to be executed after writing the data.
61	* @return {void}
62	*/
63	_write(chunk: T, encoding: BufferEncoding, callback: TransformCallback): void {
64	this._groupBy(chunk);	30✔
65	callback();	30✔
66	}
67
68
69	/**
70	* Finalize the stream by draining the buffer and pushing any remaining chunks to the stream.
71	*
72	* @param {TransformCallback} callback - The callback to be called when the stream is finalized.
73	* @return {void}
74	*/
75	_final(callback: TransformCallback): void {
76	while (this.buffer.size > 0) {	6✔
77	const groupEntry = this.buffer.entries().next().value;	12✔
78	if (groupEntry) {	12✔
79	const [groupKey, group] = groupEntry;	12✔
80	this.push(group);	12✔
81	this.buffer.delete(groupKey);	12✔
82	}
83	}
84	this.push(null);	6✔
85	callback();	6✔
86	}
87
88	/**
89	* Reading is not supported since writer finishes first.
90	*
91	* @return {void}
92	*/
93	_read(): void {
94	return;	×
95	}
96
97	/**
98	* Groups a chunk of data based on the provided groupBy function and stores it in the buffer.
99	*
100	* @param {T} chunk - The data chunk to be grouped.
101	* @return {void} This function does not return anything.
102	*/
103	private _groupBy(chunk: T): void {
104	const groupKey = this.groupBy(chunk);	30✔
105	const group = this.buffer.get(groupKey) ?? [];	30✔
106	group.push(chunk);	30✔
107	this.buffer.set(groupKey, group);	30✔
108	}
109	}

palcarazm / batchjs / 14983613232

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