28286855955

Committed 27 Jun 2026 10:43AM UTC coverage: 98.217% (-1.3%) from 99.54%

Build # 28286855955

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #107 from palcarazm/palcarazm/issue106

feat: implement Runnable base class with lifecycle events

Coverage Stats

160 of 169 branches covered (94.67%)

Branch coverage included in aggregate %.

174 of 178 new or added lines in 10 files covered. (97.75%)

611 of 616 relevant lines covered (99.19%)

98.34 hits per line

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

94.92

/src/main/common/interfaces/step/Step.ts

import { Readable, Duplex, Writable } from "node:stream";
import { Runnable , RunnableStatus } from "../runnable/_index";
import { StepEventMap } from "./StepEvents";
import { StepCancelledError } from "../../errors/_index";

/**
 * @abstract
 * @class
 * Abstract base class for all steps.
 * @extends {@link Runnable<StepEventMap>}
 * @example
 * ```typescript
 * class StepImplementation extends Step {
    constructor(name: string = "MockPassingStep") {
        super(name);
    }

    protected _reader() {
        return new Readable({
            objectMode: true,
            read() {
                this.push("data");
                this.push(null);
            }
        });
    }

    protected _processors() {
        const opts: TransformOptions = {
            objectMode: true,
            transform(chunk: unknown, encoding: BufferEncoding, callback: TransformCallback) {
                this.push(chunk);
                callback();
            }
        };
        return [new Transform(opts), new Transform(opts)];
    }

    protected _writer() {
        return new Writable({
            objectMode: true,
            write(chunk: unknown, encoding: BufferEncoding, callback: TransformCallback) {
                callback();
            }
        });
    }
}
 * const step = new StepImplementation("StepImplementation");
 * step.on("started", () => console.log("Step started"));
 * step.on("completed", () => console.log("Step completed"));
 * step.run();
 * ```
 * ```shell
 * >> Step started
 * >> Step completed
 * ```
 */
export abstract class Step extends Runnable<StepEventMap> {
    private _readerInstance?:Readable;
    private _processorsInstances?:Duplex[];
    private _writerInstance?:Writable;

    /**
     * @param {string} name - The name to assign to the Step.
     * @param {object} params - The parameters to pass to the step.
     */
    constructor(name:string,params:object={}) {
        super(name, params);
    }
    
    /**
     * @abstract
     * Abstract method that must be implemented by the step in order to defined the reader stream.
     * @returns {Readable}
     * @protected
     */
    protected abstract _reader():Readable;

    /**
     * @abstract
     * Abstract method that must be implemented by the step in order to process the data from the reader stream and push it to the writer stream.
     * Processors are defined in an ordered array to be chained on the runner.
     * @returns {Array<Duplex>}
     * @protected
     */
    protected abstract _processors():Array<Duplex>;

    /**
     * @abstract
     * Abstract method that must be implemented by the step in order to defined the writer stream.
     * @returns {Writable}
     * @protected
     */
    protected abstract _writer():Writable;

    /**
     * Hook called during transition to RUNNING.
     * Builds the stream pipeline and launches it asynchronously.
     * @returns {Promise<{ cancelled: boolean; reason?: string; executionPromise: Promise<void> }>}
     */
    protected async doRun(): Promise<{ cancelled: boolean; reason?: string, executionPromise: Promise<void> }> {
        this._readerInstance = this._reader();
        this._processorsInstances = this._processors();
        this._writerInstance = this._writer();

        const executionPromise = new Promise<void>((resolve, reject) => {
            this._readerInstance!.once("error", (error) => {
                reject(error);
            });
            this._writerInstance!.once("close", () => {
                if (this.isCancelled || this.transitioningTo === RunnableStatus.CANCELLED) {
                    reject(new StepCancelledError(this.name));
                }
            }).once("error", (error) => {
                reject(error);
            });

            let assembly: Readable = this._readerInstance!;
            for (const processor of this._processorsInstances!) {
                processor.once("error", (error) => {
                    reject(error);
                });
                assembly = assembly.pipe(processor);
            }
            assembly.pipe(this._writerInstance!);

            assembly
                .once("finish", () => {
                    resolve();
                });
        }).then(() => {
            return this.transitionTo(RunnableStatus.COMPLETED);
        }).catch((error) => {
            if (this.isRunning && this.transitioningTo === undefined) {
                return this.transitionTo(RunnableStatus.FAILED, error)
                    .finally(() => { throw error; });
            }
            throw error;
        });

        return { cancelled: false, executionPromise };
    }

    /**
     * Hook called during transition to COMPLETED.
     * Destroys all streams.
     * @returns {Promise<{ cancelled: boolean; reason?: string }>}
     */
    protected async doComplete(): Promise<{ cancelled: boolean; reason?: string }> {
        this.destroy();
        return { cancelled: false };
    }

    /**
     * Hook called during transition to FAILED.
     * Destroys all streams.
     * @param {Error} _error - The error that caused the failure.
     * @returns {Promise<{ cancelled: boolean; reason?: string }>}
     */
    // eslint-disable-next-line @typescript-eslint/no-unused-vars
    protected async doFail(_error: Error): Promise<{ cancelled: boolean; reason?: string }> {
        this.destroy();
        return { cancelled: false };
    }

    /**
     * Hook called during transition to CANCELLED.
     * Destroys all streams.
     * @returns {Promise<{ cancelled: boolean; reason?: string }>}
     */
    protected async doCancel(): Promise<{ cancelled: boolean; reason?: string }> {
        this.destroy();
        return { cancelled: false };
    }

    /**
     * Destroys all resources associated with the step.
     * Called automatically on completion, failure, or cancellation.
     * @returns {void}
     * @protected
     */
    protected destroy(): void {
        if (this.isCreated || (this.isRunning && this.transitioningTo === undefined)) {
            return;
        }
        this._readerInstance?.destroy();
        this._writerInstance?.destroy();
        for (const processor of this._processorsInstances ?? []) {
            processor.destroy();
        }
    }
}

1	import { Readable, Duplex, Writable } from "node:stream";
2	import { Runnable , RunnableStatus } from "../runnable/_index";	36✔
3	import { StepEventMap } from "./StepEvents";
4	import { StepCancelledError } from "../../errors/_index";	36✔
5
6	/**
7	* @abstract
8	* @class
9	* Abstract base class for all steps.
10	* @extends {@link Runnable<StepEventMap>}
11	* @example
12	* ```typescript
13	* class StepImplementation extends Step {
14	constructor(name: string = "MockPassingStep") {
15	super(name);
16	}
17
18	protected _reader() {
19	return new Readable({
20	objectMode: true,
21	read() {
22	this.push("data");
23	this.push(null);
24	}
25	});
26	}
27
28	protected _processors() {
29	const opts: TransformOptions = {
30	objectMode: true,
31	transform(chunk: unknown, encoding: BufferEncoding, callback: TransformCallback) {
32	this.push(chunk);
33	callback();
34	}
35	};
36	return [new Transform(opts), new Transform(opts)];
37	}
38
39	protected _writer() {
40	return new Writable({
41	objectMode: true,
42	write(chunk: unknown, encoding: BufferEncoding, callback: TransformCallback) {
43	callback();
44	}
45	});
46	}
47	}
48	* const step = new StepImplementation("StepImplementation");
49	* step.on("started", () => console.log("Step started"));
50	* step.on("completed", () => console.log("Step completed"));
51	* step.run();
52	* ```
53	* ```shell
54	* >> Step started
55	* >> Step completed
56	* ```
57	*/
58	export abstract class Step extends Runnable<StepEventMap> {	36✔
59	private _readerInstance?:Readable;
60	private _processorsInstances?:Duplex[];
61	private _writerInstance?:Writable;
62
63	/**
64	* @param {string} name - The name to assign to the Step.
65	* @param {object} params - The parameters to pass to the step.
66	*/
67	constructor(name:string,params:object={}) {	142✔
68	super(name, params);	328✔
69	}
70
71	/**
72	* @abstract
73	* Abstract method that must be implemented by the step in order to defined the reader stream.
74	* @returns {Readable}
75	* @protected
76	*/
77	protected abstract _reader():Readable;
78
79	/**
80	* @abstract
81	* Abstract method that must be implemented by the step in order to process the data from the reader stream and push it to the writer stream.
82	* Processors are defined in an ordered array to be chained on the runner.
83	* @returns {Array<Duplex>}
84	* @protected
85	*/
86	protected abstract _processors():Array<Duplex>;
87
88	/**
89	* @abstract
90	* Abstract method that must be implemented by the step in order to defined the writer stream.
91	* @returns {Writable}
92	* @protected
93	*/
94	protected abstract _writer():Writable;
95
96	/**
97	* Hook called during transition to RUNNING.
98	* Builds the stream pipeline and launches it asynchronously.
99	* @returns {Promise<{ cancelled: boolean; reason?: string; executionPromise: Promise<void> }>}
100	*/
101	protected async doRun(): Promise<{ cancelled: boolean; reason?: string, executionPromise: Promise<void> }> {
102	this._readerInstance = this._reader();	228✔
103	this._processorsInstances = this._processors();	228✔
104	this._writerInstance = this._writer();	228✔
105
106	const executionPromise = new Promise<void>((resolve, reject) => {	228✔
107	this._readerInstance!.once("error", (error) => {	228✔
108	reject(error);	16✔
109	});
110	this._writerInstance!.once("close", () => {	228✔
111	if (this.isCancelled \|\| this.transitioningTo === RunnableStatus.CANCELLED) {	204✔
112	reject(new StepCancelledError(this.name));	48✔
113	}
114	}).once("error", (error) => {
115	reject(error);	4✔
116	});
117
118	let assembly: Readable = this._readerInstance!;	228✔
119	for (const processor of this._processorsInstances!) {	228✔
120	processor.once("error", (error) => {	412✔
121	reject(error);	36✔
122	});
123	assembly = assembly.pipe(processor);	412✔
124	}
125	assembly.pipe(this._writerInstance!);	228✔
126
127	assembly	228✔
128	.once("finish", () => {
129	resolve();	116✔
130	});
131	}).then(() => {
132	return this.transitionTo(RunnableStatus.COMPLETED);	112✔
133	}).catch((error) => {
134	if (this.isRunning && this.transitioningTo === undefined) {	104✔
135	return this.transitionTo(RunnableStatus.FAILED, error)	56✔
136	.finally(() => { throw error; });	56✔
137	}
138	throw error;	48✔
139	});
140
141	return { cancelled: false, executionPromise };	228✔
142	}
143
144	/**
145	* Hook called during transition to COMPLETED.
146	* Destroys all streams.
147	* @returns {Promise<{ cancelled: boolean; reason?: string }>}
148	*/
149	protected async doComplete(): Promise<{ cancelled: boolean; reason?: string }> {
150	this.destroy();	112✔
151	return { cancelled: false };	112✔
152	}
153
154	/**
155	* Hook called during transition to FAILED.
156	* Destroys all streams.
157	* @param {Error} _error - The error that caused the failure.
158	* @returns {Promise<{ cancelled: boolean; reason?: string }>}
159	*/
160	// eslint-disable-next-line @typescript-eslint/no-unused-vars
161	protected async doFail(_error: Error): Promise<{ cancelled: boolean; reason?: string }> {
162	this.destroy();	56✔
163	return { cancelled: false };	56✔
164	}
165
166	/**
167	* Hook called during transition to CANCELLED.
168	* Destroys all streams.
169	* @returns {Promise<{ cancelled: boolean; reason?: string }>}
170	*/
171	protected async doCancel(): Promise<{ cancelled: boolean; reason?: string }> {
172	this.destroy();	60✔
173	return { cancelled: false };	60✔
174	}
175
176	/**
177	* Destroys all resources associated with the step.
178	* Called automatically on completion, failure, or cancellation.
179	* @returns {void}
180	* @protected
181	*/
182	protected destroy(): void {
183	if (this.isCreated \|\| (this.isRunning && this.transitioningTo === undefined)) {	228!
NEW 184	return;	×
185	}
186	this._readerInstance?.destroy();	228✔
187	this._writerInstance?.destroy();	228✔
188	for (const processor of this._processorsInstances ?? []) {	228!
189	processor.destroy();	412✔
190	}
191	}
192	}

palcarazm / batchjs / 28286855955

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