10223129684

Committed 02 Aug 2024 11:08PM UTC coverage: 94.639%. First build

Build # 10223129684

Build Type

Pull #21

github

Committed by

snatalenko

Commit Message

Separate github workflows for tests and coveralls

Pull Request Pull Request #21: Migrate to TypeScript

Run Details

552 of 854 branches covered (64.64%)

2231 of 2360 new or added lines in 28 files covered. (94.53%)

2348 of 2481 relevant lines covered (94.64%)

21.9 hits per line

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

97.67

/src/AggregateCommandHandler.ts

import {
        IAggregate,
        IAggregateConstructor,
        IAggregateFactory,
        ICommand,
        ICommandBus,
        ICommandHandler,
        Identifier,
        IEventSet,
        IEventStore,
        IExtendableLogger,
        ILogger
} from "./interfaces";

import {
        getClassName,
        getHandledMessageTypes,
        subscribe
} from './utils';

/**
 * Aggregate command handler.
 *
 * Subscribes to event store and awaits aggregate commands.
 * Upon command receiving creates an instance of aggregate,
 * restores its state, passes command and commits emitted events to event store.
 */
export class AggregateCommandHandler implements ICommandHandler {

        #eventStore: IEventStore;
        #logger?: ILogger;

        #aggregateFactory: IAggregateFactory<any>;
        #handles: string[];

        constructor({
                eventStore,
                aggregateType,
                aggregateFactory,
                handles,
                logger
        }: {
                eventStore: IEventStore,
                aggregateType?: IAggregateConstructor<any>,
                aggregateFactory?: IAggregateFactory<any>,
                handles?: string[],
                logger?: ILogger | IExtendableLogger
        }) {
                if (!eventStore)
                        throw new TypeError('eventStore argument required');

                this.#eventStore = eventStore;
                this.#logger = logger && 'child' in logger ?
                        logger.child({ service: getClassName(this) }) :
                        logger;

                if (aggregateType) {
                        const AggregateType = aggregateType;
                        this.#aggregateFactory = params => new AggregateType(params);
                        this.#handles = getHandledMessageTypes(AggregateType);
                }
                else if (aggregateFactory) {
                        if (!Array.isArray(handles) || !handles.length)
                                throw new TypeError('handles argument must be an non-empty Array');

                        this.#aggregateFactory = aggregateFactory;
                        this.#handles = handles;
                }
                else {
                        throw new TypeError('either aggregateType or aggregateFactory is required');
                }
        }

        /** Subscribe to all command types handled by aggregateType */
        subscribe(commandBus: ICommandBus) {
                subscribe(commandBus, this, {
                        messageTypes: this.#handles,
                        masterHandler: (c: ICommand) => this.execute(c)
                });
        }

        /** Restore aggregate from event store events */
        async #restoreAggregate(id: Identifier): Promise<IAggregate> {
                if (!id)
                        throw new TypeError('id argument required');

                const events = await this.#eventStore.getAggregateEvents(id);
                const aggregate = this.#aggregateFactory({ id, events });

                this.#logger?.info(`${aggregate} state restored from ${events.length} event(s)`);

                return aggregate;
        }

        /** Create new aggregate with new Id generated by event store */
        async #createAggregate(): Promise<IAggregate> {
                const id = await this.#eventStore.getNewId();
                const aggregate = this.#aggregateFactory({ id });
                this.#logger?.info(`${aggregate} created`);

                return aggregate;
        }

        /** Pass a command to corresponding aggregate */
        async execute(cmd: ICommand): Promise<IEventSet> {
                if (!cmd) throw new TypeError('cmd argument required');
                if (!cmd.type) throw new TypeError('cmd.type argument required');

                const aggregate = cmd.aggregateId ?
                        await this.#restoreAggregate(cmd.aggregateId) :
                        await this.#createAggregate();

                await aggregate.handle(cmd);

                let events = aggregate.changes;
                this.#logger?.info(`${aggregate} "${cmd.type}" command processed, ${events.length} event(s) produced`);
                if (!events.length)
                        return events;

                if (aggregate.shouldTakeSnapshot && this.#eventStore.snapshotsSupported) {
                        aggregate.takeSnapshot();
                        events = aggregate.changes;
                }

                await this.#eventStore.commit(events);

                return events;
        }
}

1	import {	4✔
2	IAggregate,	4✔
3	IAggregateConstructor,	4✔
4	IAggregateFactory,	4✔
5	ICommand,	4✔
6	ICommandBus,	4✔
7	ICommandHandler,	4✔
8	Identifier,	4✔
9	IEventSet,	4✔
10	IEventStore,	4✔
11	IExtendableLogger,	4✔
12	ILogger	4✔
13	} from "./interfaces";	4✔
14		4✔
15	import {	4✔
16	getClassName,	4✔
17	getHandledMessageTypes,	4✔
18	subscribe	4✔
19	} from './utils';	4✔
20		4✔
21	/**	4✔
22	* Aggregate command handler.	4✔
23	*	4✔
24	* Subscribes to event store and awaits aggregate commands.	4✔
25	* Upon command receiving creates an instance of aggregate,	4✔
26	* restores its state, passes command and commits emitted events to event store.	4✔
27	*/	4✔
28	export class AggregateCommandHandler implements ICommandHandler {	16✔
29		16✔
30	#eventStore: IEventStore;	16✔
31	#logger?: ILogger;	16✔
32		16✔
33	#aggregateFactory: IAggregateFactory<any>;	16✔
34	#handles: string[];	16✔
35		16✔
36	constructor({	16✔
37	eventStore,	16✔
38	aggregateType,	16✔
39	aggregateFactory,	16✔
40	handles,	16✔
41	logger	16✔
42	}: {	16✔
43	eventStore: IEventStore,	16✔
44	aggregateType?: IAggregateConstructor<any>,	16✔
45	aggregateFactory?: IAggregateFactory<any>,	16✔
46	handles?: string[],	16✔
47	logger?: ILogger \| IExtendableLogger	16✔
48	}) {	16✔
49	if (!eventStore)	16✔
50	throw new TypeError('eventStore argument required');	16!
51		16✔
52	this.#eventStore = eventStore;	16✔
53	this.#logger = logger && 'child' in logger ?	16!
54	logger.child({ service: getClassName(this) }) :	16✔
55	logger;	16✔
56		16✔
57	if (aggregateType) {	16✔
58	const AggregateType = aggregateType;	12✔
59	this.#aggregateFactory = params => new AggregateType(params);	12✔
60	this.#handles = getHandledMessageTypes(AggregateType);	12✔
61	}	12✔
62	else if (aggregateFactory) {	4✔
63	if (!Array.isArray(handles) \|\| !handles.length)	4✔
64	throw new TypeError('handles argument must be an non-empty Array');	4!
65		4✔
66	this.#aggregateFactory = aggregateFactory;	4✔
67	this.#handles = handles;	4✔
68	}	4✔
NEW 69	else {	×
NEW 70	throw new TypeError('either aggregateType or aggregateFactory is required');	×
NEW 71	}	×
72	}	16✔
73		16✔
74	/** Subscribe to all command types handled by aggregateType */	16✔
75	subscribe(commandBus: ICommandBus) {	16✔
76	subscribe(commandBus, this, {	2✔
77	messageTypes: this.#handles,	2✔
78	masterHandler: (c: ICommand) => this.execute(c)	2✔
79	});	2✔
80	}	2✔
81		16✔
82	/** Restore aggregate from event store events */	16✔
83	async #restoreAggregate(id: Identifier): Promise<IAggregate> {	16✔
84	if (!id)	6✔
85	throw new TypeError('id argument required');	6!
86		6✔
87	const events = await this.#eventStore.getAggregateEvents(id);	6✔
88	const aggregate = this.#aggregateFactory({ id, events });	6✔
89		6✔
90	this.#logger?.info(`${aggregate} state restored from ${events.length} event(s)`);	6!
91		6✔
92	return aggregate;	6✔
93	}	6✔
94		16✔
95	/** Create new aggregate with new Id generated by event store */	16✔
96	async #createAggregate(): Promise<IAggregate> {	16✔
97	const id = await this.#eventStore.getNewId();	10✔
98	const aggregate = this.#aggregateFactory({ id });	10✔
99	this.#logger?.info(`${aggregate} created`);	10!
100		10✔
101	return aggregate;	10✔
102	}	10✔
103		16✔
104	/** Pass a command to corresponding aggregate */	16✔
105	async execute(cmd: ICommand): Promise<IEventSet> {	16✔
106	if (!cmd) throw new TypeError('cmd argument required');	16!
107	if (!cmd.type) throw new TypeError('cmd.type argument required');	16!
108		16✔
109	const aggregate = cmd.aggregateId ?	16✔
110	await this.#restoreAggregate(cmd.aggregateId) :	16✔
111	await this.#createAggregate();	16✔
112		10✔
113	await aggregate.handle(cmd);	10✔
114		16✔
115	let events = aggregate.changes;	16✔
116	this.#logger?.info(`${aggregate} "${cmd.type}" command processed, ${events.length} event(s) produced`);	16!
117	if (!events.length)	16✔
118	return events;	16!
119		16✔
120	if (aggregate.shouldTakeSnapshot && this.#eventStore.snapshotsSupported) {	16✔
121	aggregate.takeSnapshot();	2✔
122	events = aggregate.changes;	2✔
123	}	2✔
124		16✔
125	await this.#eventStore.commit(events);	16✔
126		16✔
127	return events;	16✔
128	}	16✔
129	}	16✔

snatalenko / node-cqrs / 10223129684

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