7341002981

Committed 27 Dec 2023 06:37PM UTC coverage: 96.351% (+0.3%) from 96.063%

Build # 7341002981

Build Type

push

github

Committed by

web-flow

Commit Message

feat!: Drop support for unsupported versions of Node.js (#833)

* feat!: Drop support for unsupported versions of Node.js

* Upgrade requisite libraries for node 20

* Remove whitesource unified agent

* Run npm audit fix

* Fix lint arrow body style

* Fix CI

Run Details

225 of 244 branches covered (0.0%)

Branch coverage included in aggregate %.

45 of 45 new or added lines in 9 files covered. (100.0%)

2 existing lines in 1 file now uncovered.

752 of 770 relevant lines covered (97.66%)

1089511.36 hits per line

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

92.81

/packages/parse/src/CsvParserStream.ts

import { StringDecoder } from 'string_decoder';
import { Transform, TransformCallback } from 'stream';
import { ParserOptions } from './ParserOptions';
import { HeaderTransformer, RowTransformerValidator } from './transforms';
import { Parser } from './parser';
import { Row, RowArray, RowTransformFunction, RowValidate, RowValidatorCallback } from './types';

export class CsvParserStream<I extends Row, O extends Row> extends Transform {
    private readonly parserOptions: ParserOptions;

    private readonly decoder: StringDecoder;

    private readonly parser: Parser;

    private readonly headerTransformer: HeaderTransformer<I>;

    private readonly rowTransformerValidator: RowTransformerValidator<I, O>;

    private lines = '';

    private rowCount = 0;

    private parsedRowCount = 0;

    private parsedLineCount = 0;

    private endEmitted = false;

    private headersEmitted = false;

    public constructor(parserOptions: ParserOptions) {
        super({ objectMode: parserOptions.objectMode });
        this.parserOptions = parserOptions;
        this.parser = new Parser(parserOptions);
        this.headerTransformer = new HeaderTransformer(parserOptions);
        this.decoder = new StringDecoder(parserOptions.encoding);
        this.rowTransformerValidator = new RowTransformerValidator();
    }

    private get hasHitRowLimit(): boolean {
        return this.parserOptions.limitRows && this.rowCount >= this.parserOptions.maxRows;
    }

    private get shouldEmitRows(): boolean {
        return this.parsedRowCount > this.parserOptions.skipRows;
    }

    private get shouldSkipLine(): boolean {
        return this.parsedLineCount <= this.parserOptions.skipLines;
    }

    public transform(transformFunction: RowTransformFunction<I, O>): CsvParserStream<I, O> {
        this.rowTransformerValidator.rowTransform = transformFunction;
        return this;
    }

    public validate(validateFunction: RowValidate<O>): CsvParserStream<I, O> {
        this.rowTransformerValidator.rowValidator = validateFunction;
        return this;
    }

    // eslint-disable-next-line @typescript-eslint/no-explicit-any
    public emit(event: string | symbol, ...rest: any[]): boolean {
        if (event === 'end') {
            if (!this.endEmitted) {
                this.endEmitted = true;
                super.emit('end', this.rowCount);
            }
            return false;
        }
        return super.emit(event, ...rest);
    }

    public _transform(data: Buffer, encoding: string, done: TransformCallback): void {
        // if we have hit our maxRows parsing limit then skip parsing
        if (this.hasHitRowLimit) {
            return done();
        }
        const wrappedCallback = CsvParserStream.wrapDoneCallback(done);
        try {
            const { lines } = this;
            const newLine = lines + this.decoder.write(data);
            const rows = this.parse(newLine, true);
            return this.processRows(rows, wrappedCallback);
        } catch (e) {
            return wrappedCallback(e);
        }
    }

    public _flush(done: TransformCallback): void {
        const wrappedCallback = CsvParserStream.wrapDoneCallback(done);
        // if we have hit our maxRows parsing limit then skip parsing
        if (this.hasHitRowLimit) {
            return wrappedCallback();
        }
        try {
            const newLine = this.lines + this.decoder.end();
            const rows = this.parse(newLine, false);
            return this.processRows(rows, wrappedCallback);
        } catch (e) {
            return wrappedCallback(e);
        }
    }

    private parse(data: string, hasMoreData: boolean): string[][] {
        if (!data) {
            return [];
        }
        const { line, rows } = this.parser.parse(data, hasMoreData);
        this.lines = line;
        return rows;
    }

    private processRows(rows: string[][], cb: TransformCallback): void {
        const rowsLength = rows.length;
        const iterate = (i: number): void => {
            const callNext = (err?: Error): void => {
                if (err) {
                    return cb(err);
                }
                if (i % 100 === 0) {
                    // incase the transform are sync insert a next tick to prevent stack overflow
                    setImmediate((): void => {
                        return iterate(i + 1);
                    });
                    return undefined;
                }
                return iterate(i + 1);
            };
            this.checkAndEmitHeaders();
            // if we have emitted all rows or we have hit the maxRows limit option
            // then end
            if (i >= rowsLength || this.hasHitRowLimit) {
                return cb();
            }
            this.parsedLineCount += 1;
            if (this.shouldSkipLine) {
                return callNext();
            }
            const row = rows[i];
            this.rowCount += 1;
            this.parsedRowCount += 1;
            const nextRowCount = this.rowCount;
            return this.transformRow(row, (err, transformResult): void => {
                if (err) {
                    this.rowCount -= 1;
                    return callNext(err);
                }
                if (!transformResult) {
                    return callNext(new Error('expected transform result'));
                }
                if (!transformResult.isValid) {
                    this.emit('data-invalid', transformResult.row, nextRowCount, transformResult.reason);
                } else if (transformResult.row) {
                    return this.pushRow(transformResult.row, callNext);
                }
                return callNext();
            });
        };
        iterate(0);
    }

    private transformRow(parsedRow: RowArray, cb: RowValidatorCallback<O>): void {
        try {
            this.headerTransformer.transform(parsedRow, (err, withHeaders): void => {
                if (err) {
                    return cb(err);
                }
                if (!withHeaders) {
                    return cb(new Error('Expected result from header transform'));
                }
                if (!withHeaders.isValid) {
                    if (this.shouldEmitRows) {
                        return cb(null, { isValid: false, row: parsedRow as never as O });
                    }
                    // skipped because of skipRows option remove from total row count
                    return this.skipRow(cb);
                }
                if (withHeaders.row) {
                    if (this.shouldEmitRows) {
                        return this.rowTransformerValidator.transformAndValidate(withHeaders.row, cb);
                    }
                    // skipped because of skipRows option remove from total row count
                    return this.skipRow(cb);
                }
                // this is a header row dont include in the rowCount or parsedRowCount
                this.rowCount -= 1;
                this.parsedRowCount -= 1;
                return cb(null, { row: null, isValid: true });
            });
        } catch (e) {
            cb(e);
        }
    }

    private checkAndEmitHeaders(): void {
        if (!this.headersEmitted && this.headerTransformer.headers) {
            this.headersEmitted = true;
            this.emit('headers', this.headerTransformer.headers);
        }
    }

    private skipRow(cb: RowValidatorCallback<O>): void {
        // skipped because of skipRows option remove from total row count
        this.rowCount -= 1;
        return cb(null, { row: null, isValid: true });
    }

    private pushRow(row: Row, cb: (err?: Error) => void): void {
        try {
            if (!this.parserOptions.objectMode) {
                this.push(JSON.stringify(row));
            } else {
                this.push(row);
            }
            cb();
        } catch (e) {
            cb(e);
        }
    }

    private static wrapDoneCallback(done: TransformCallback): TransformCallback {
        let errorCalled = false;
        // eslint-disable-next-line @typescript-eslint/no-explicit-any
        return (err: Error | null | undefined, ...args: any[]): void => {
            if (err) {
                if (errorCalled) {
                    throw err;
                }
                errorCalled = true;
                done(err);
                return;
            }
            done(...args);
        };
    }
}

1	import { StringDecoder } from 'string_decoder';	60✔
2	import { Transform, TransformCallback } from 'stream';	60✔
3	import { ParserOptions } from './ParserOptions';
4	import { HeaderTransformer, RowTransformerValidator } from './transforms';	60✔
5	import { Parser } from './parser';	60✔
6	import { Row, RowArray, RowTransformFunction, RowValidate, RowValidatorCallback } from './types';
7
8	export class CsvParserStream<I extends Row, O extends Row> extends Transform {	60✔
9	private readonly parserOptions: ParserOptions;
10
11	private readonly decoder: StringDecoder;
12
13	private readonly parser: Parser;
14
15	private readonly headerTransformer: HeaderTransformer<I>;
16
17	private readonly rowTransformerValidator: RowTransformerValidator<I, O>;
18
19	private lines = '';	234✔
20
21	private rowCount = 0;	234✔
22
23	private parsedRowCount = 0;	234✔
24
25	private parsedLineCount = 0;	234✔
26
27	private endEmitted = false;	234✔
28
29	private headersEmitted = false;	234✔
30
31	public constructor(parserOptions: ParserOptions) {
32	super({ objectMode: parserOptions.objectMode });	234✔
33	this.parserOptions = parserOptions;	234✔
34	this.parser = new Parser(parserOptions);	234✔
35	this.headerTransformer = new HeaderTransformer(parserOptions);	234✔
36	this.decoder = new StringDecoder(parserOptions.encoding);	234✔
37	this.rowTransformerValidator = new RowTransformerValidator();	234✔
38	}
39
40	private get hasHitRowLimit(): boolean {
41	return this.parserOptions.limitRows && this.rowCount >= this.parserOptions.maxRows;	365,493✔
42	}
43
44	private get shouldEmitRows(): boolean {
45	return this.parsedRowCount > this.parserOptions.skipRows;	364,188✔
46	}
47
48	private get shouldSkipLine(): boolean {
49	return this.parsedLineCount <= this.parserOptions.skipLines;	364,404✔
50	}
51
52	public transform(transformFunction: RowTransformFunction<I, O>): CsvParserStream<I, O> {
53	this.rowTransformerValidator.rowTransform = transformFunction;	30✔
54	return this;	27✔
55	}
56
57	public validate(validateFunction: RowValidate<O>): CsvParserStream<I, O> {
58	this.rowTransformerValidator.rowValidator = validateFunction;	27✔
59	return this;	24✔
60	}
61
62	// eslint-disable-next-line @typescript-eslint/no-explicit-any
63	public emit(event: string \| symbol, ...rest: any[]): boolean {
64	if (event === 'end') {	365,739✔
65	if (!this.endEmitted) {	180✔
66	this.endEmitted = true;	180✔
67	super.emit('end', this.rowCount);	180✔
68	}
69	return false;	171✔
70	}
71	return super.emit(event, ...rest);	365,559✔
72	}
73
74	public _transform(data: Buffer, encoding: string, done: TransformCallback): void {
75	// if we have hit our maxRows parsing limit then skip parsing
76	if (this.hasHitRowLimit) {	891!
77	return done();	×
78	}
79	const wrappedCallback = CsvParserStream.wrapDoneCallback(done);	891✔
80	try {	891✔
81	const { lines } = this;	891✔
82	const newLine = lines + this.decoder.write(data);	891✔
83	const rows = this.parse(newLine, true);	891✔
84	return this.processRows(rows, wrappedCallback);	885✔
85	} catch (e) {
86	return wrappedCallback(e);	6✔
87	}
88	}
89
90	public _flush(done: TransformCallback): void {
91	const wrappedCallback = CsvParserStream.wrapDoneCallback(done);	195✔
92	// if we have hit our maxRows parsing limit then skip parsing
93	if (this.hasHitRowLimit) {	195✔
94	return wrappedCallback();	3✔
95	}
96	try {	192✔
97	const newLine = this.lines + this.decoder.end();	192✔
98	const rows = this.parse(newLine, false);	192✔
99	return this.processRows(rows, wrappedCallback);	192✔
100	} catch (e) {
UNCOV 101	return wrappedCallback(e);	×
102	}
103	}
104
105	private parse(data: string, hasMoreData: boolean): string[][] {
106	if (!data) {	1,083✔
107	return [];	3✔
108	}
109	const { line, rows } = this.parser.parse(data, hasMoreData);	1,080✔
110	this.lines = line;	1,074✔
111	return rows;	1,074✔
112	}
113
114	private processRows(rows: string[][], cb: TransformCallback): void {
115	const rowsLength = rows.length;	1,077✔
116	const iterate = (i: number): void => {	1,077✔
117	const callNext = (err?: Error): void => {	365,442✔
118	if (err) {	364,404✔
119	return cb(err);	39✔
120	}
121	if (i % 100 === 0) {	364,365✔
122	// incase the transform are sync insert a next tick to prevent stack overflow
123	setImmediate((): void => {	4,152✔
124	return iterate(i + 1);	4,152✔
125	});
126	return undefined;	4,152✔
127	}
128	return iterate(i + 1);	360,213✔
129	};
130	this.checkAndEmitHeaders();	365,442✔
131	// if we have emitted all rows or we have hit the maxRows limit option
132	// then end
133	if (i >= rowsLength \|\| this.hasHitRowLimit) {	365,442✔
134	return cb();	1,038✔
135	}
136	this.parsedLineCount += 1;	364,404✔
137	if (this.shouldSkipLine) {	364,404✔
138	return callNext();	24✔
139	}
140	const row = rows[i];	364,380✔
141	this.rowCount += 1;	364,380✔
142	this.parsedRowCount += 1;	364,380✔
143	const nextRowCount = this.rowCount;	364,380✔
144	return this.transformRow(row, (err, transformResult): void => {	364,380✔
145	if (err) {	364,380✔
146	this.rowCount -= 1;	36✔
147	return callNext(err);	36✔
148	}
149	if (!transformResult) {	364,344!
150	return callNext(new Error('expected transform result'));	×
151	}
152	if (!transformResult.isValid) {	364,344✔
153	this.emit('data-invalid', transformResult.row, nextRowCount, transformResult.reason);	51✔
154	} else if (transformResult.row) {	364,293✔
155	return this.pushRow(transformResult.row, callNext);	364,047✔
156	}
157	return callNext();	297✔
158	});
159	};
160	iterate(0);	1,077✔
161	}
162
163	private transformRow(parsedRow: RowArray, cb: RowValidatorCallback<O>): void {
164	try {	364,380✔
165	this.headerTransformer.transform(parsedRow, (err, withHeaders): void => {	364,380✔
166	if (err) {	364,368!
167	return cb(err);	×
168	}
169	if (!withHeaders) {	364,368!
170	return cb(new Error('Expected result from header transform'));	×
171	}
172	if (!withHeaders.isValid) {	364,368✔
173	if (this.shouldEmitRows) {	12✔
174	return cb(null, { isValid: false, row: parsedRow as never as O });	6✔
175	}
176	// skipped because of skipRows option remove from total row count
177	return this.skipRow(cb);	6✔
178	}
179	if (withHeaders.row) {	364,356✔
180	if (this.shouldEmitRows) {	364,176✔
181	return this.rowTransformerValidator.transformAndValidate(withHeaders.row, cb);	364,116✔
182	}
183	// skipped because of skipRows option remove from total row count
184	return this.skipRow(cb);	60✔
185	}
186	// this is a header row dont include in the rowCount or parsedRowCount
187	this.rowCount -= 1;	180✔
188	this.parsedRowCount -= 1;	180✔
189	return cb(null, { row: null, isValid: true });	180✔
190	});
191	} catch (e) {
192	cb(e);	24✔
193	}
194	}
195
196	private checkAndEmitHeaders(): void {
197	if (!this.headersEmitted && this.headerTransformer.headers) {	365,442✔
198	this.headersEmitted = true;	192✔
199	this.emit('headers', this.headerTransformer.headers);	192✔
200	}
201	}
202
203	private skipRow(cb: RowValidatorCallback<O>): void {
204	// skipped because of skipRows option remove from total row count
205	this.rowCount -= 1;	66✔
206	return cb(null, { row: null, isValid: true });	66✔
207	}
208
209	private pushRow(row: Row, cb: (err?: Error) => void): void {
210	try {	364,047✔
211	if (!this.parserOptions.objectMode) {	364,047✔
212	this.push(JSON.stringify(row));	27✔
213	} else {
214	this.push(row);	364,020✔
215	}
216	cb();	364,044✔
217	} catch (e) {
218	cb(e);	3✔
219	}
220	}
221
222	private static wrapDoneCallback(done: TransformCallback): TransformCallback {
223	let errorCalled = false;	1,086✔
224	// eslint-disable-next-line @typescript-eslint/no-explicit-any
225	return (err: Error \| null \| undefined, ...args: any[]): void => {	1,086✔
226	if (err) {	1,086✔
227	if (errorCalled) {	45!
UNCOV 228	throw err;	×
229	}
230	errorCalled = true;	45✔
231	done(err);	45✔
232	return;	45✔
233	}
234	done(...args);	1,041✔
235	};
236	}
237	}

C2FO / fast-csv / 7341002981

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