29819832096

Committed 21 Jul 2026 09:49AM UTC coverage: 94.047% (+0.3%) from 93.705%

Build # 29819832096

Build Type

Pull #1342

github

Committed by

web-flow

Commit Message

Merge 2660dc9cb into c32e3eff8

Pull Request Pull Request #1342: Forward termination signals to cluster workers in runner

Coverage Stats

5339 of 5898 branches covered (90.52%)

Branch coverage included in aggregate %.

7221 of 7457 relevant lines covered (96.84%)

515.86 hits per line

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

95.58

/src/registry/node-catalog.js

/*
 * moleculer
 * Copyright (c) 2023 MoleculerJS (https://github.com/moleculerjs/moleculer)
 * MIT Licensed
 */

"use strict";

const _ = require("lodash");
const os = require("os");
const Node = require("./node");
const { getIpList } = require("../utils");

/**
 * Import types
 *
 * @typedef {import("./registry")} Registry
 * @typedef {import("../service-broker")} ServiceBroker
 * @typedef {import("./node-catalog")} NodeCatalogClass
 * @typedef {import("./node-catalog").NodeCatalogListOptions} NodeCatalogListOptions
 * @typedef {import("./node-catalog").NodeCatalogListResult} NodeCatalogListResult
 */

/**
 * Catalog for nodes
 *
 * @class NodeCatalog
 * @implements {NodeCatalogClass}
 */
class NodeCatalog {
        /**
         * Creates an instance of NodeCatalog.
         *
         * @param {Registry} registry
         * @param {ServiceBroker} broker
         *
         * @memberof NodeCatalog
         */
        constructor(registry, broker) {
                this.registry = registry;
                this.broker = broker;
                this.logger = registry.logger;

                this.localNode = null;
                this.nodes = new Map();

                this.createLocalNode();
        }

        /**
         * Create local node with local information
         *
         * @returns
         * @memberof NodeCatalog
         */
        createLocalNode() {
                const node = new Node(this.broker.nodeID);
                node.local = true;
                node.ipList = getIpList();
                node.instanceID = this.broker.instanceID;
                node.hostname = os.hostname();
                node.client = {
                        type: "nodejs",
                        version: this.broker.MOLECULER_VERSION,
                        langVersion: process.version
                };
                node.metadata = this.broker.metadata;
                node.seq = 1;

                this.add(node.id, node);

                this.localNode = node;
                return node;
        }

        /**
         * Add a new node
         *
         * @param {String} id
         * @param {Node} node
         * @memberof NodeCatalog
         */
        add(id, node) {
                this.nodes.set(id, node);
        }

        /**
         * Check a node exist by nodeID
         *
         * @param {String} id
         * @returns
         * @memberof NodeCatalog
         */
        has(id) {
                return this.nodes.has(id);
        }

        /**
         * Get a node by nodeID
         *
         * @param {String} id
         * @returns {Node}
         * @memberof NodeCatalog
         */
        get(id) {
                return this.nodes.get(id);
        }

        /**
         * Delete a node by nodeID
         *
         * @param {String} id
         * @returns
         * @memberof NodeCatalog
         */
        delete(id) {
                return this.nodes.delete(id);
        }

        /**
         * Get count of all registered nodes
         */
        count() {
                return this.nodes.size;
        }

        /**
         * Get count of online nodes
         */
        onlineCount() {
                let count = 0;
                this.nodes.forEach(node => {
                        if (node.available) count++;
                });

                return count;
        }

        /**
         * Process incoming INFO packet payload
         *
         * @param {any} payload
         * @returns {Node}
         * @memberof NodeCatalog
         */
        processNodeInfo(payload) {
                const nodeID = payload.sender;
                //let oldNode;
                let node = this.get(nodeID);
                let isNew = false;
                let isReconnected = false;

                if (!node) {
                        isNew = true;
                        node = new Node(nodeID);

                        this.add(nodeID, node);
                } else if (!node.available) {
                        isReconnected = true;
                        node.lastHeartbeatTime = Math.round(process.uptime());
                        node.available = true;
                        node.offlineSince = null;
                }

                // Update instance
                const needRegister = node.update(payload, isReconnected);

                // Refresh services if 'seq' is greater or it is a reconnected node
                if (needRegister && node.services) {
                        this.registry.registerServices(node, node.services);
                }

                // Local notifications
                if (isNew) {
                        this.broker.broadcastLocal("$node.connected", { node, reconnected: false });
                        this.logger.info(`Node '${nodeID}' connected.`);
                        this.registry.updateMetrics();
                } else if (isReconnected) {
                        this.broker.broadcastLocal("$node.connected", { node, reconnected: true });
                        this.logger.info(`Node '${nodeID}' reconnected.`);
                        this.registry.updateMetrics();
                } else {
                        this.broker.broadcastLocal("$node.updated", { node });
                        this.logger.debug(`Node '${nodeID}' updated.`);
                }

                return node;
        }

        /**
         * Disconnected a node
         *
         * @param {String} nodeID
         * @param {Boolean} isUnexpected
         * @memberof NodeCatalog
         */
        disconnected(nodeID, isUnexpected) {
                let node = this.get(nodeID);
                if (node && node.available) {
                        node.disconnected(isUnexpected);

                        this.registry.unregisterServicesByNode(node.id);

                        this.broker.broadcastLocal("$node.disconnected", { node, unexpected: !!isUnexpected });

                        this.broker.servicesChanged(false);

                        this.registry.updateMetrics();

                        if (isUnexpected) this.logger.warn(`Node '${node.id}' disconnected unexpectedly.`);
                        else this.logger.info(`Node '${node.id}' disconnected.`);

                        if (this.broker.transit) this.broker.transit.removePendingRequestByNodeID(nodeID);
                }
        }

        /**
         * Get a node list
         *
         * @param {NodeCatalogListOptions} opts
         * @returns {NodeCatalogListResult[]}
         * @memberof NodeCatalog
         */
        list({ onlyAvailable = false, withServices = false } = {}) {
                let res = [];
                this.nodes.forEach(node => {
                        if (onlyAvailable && !node.available) return;

                        if (withServices) res.push(_.omit(node, ["rawInfo"]));
                        else res.push(_.omit(node, ["services", "rawInfo"]));
                });

                return res;
        }

        /**
         * Get a copy from node list.
         */
        toArray() {
                return Array.from(this.nodes.values());
        }
}

module.exports = NodeCatalog;

1	/*
2	* moleculer
3	* Copyright (c) 2023 MoleculerJS (https://github.com/moleculerjs/moleculer)
4	* MIT Licensed
5	*/
6
7	"use strict";
8
9	const _ = require("lodash");	200✔
10	const os = require("os");	200✔
11	const Node = require("./node");	200✔
12	const { getIpList } = require("../utils");	200✔
13
14	/**
15	* Import types
16	*
17	* @typedef {import("./registry")} Registry
18	* @typedef {import("../service-broker")} ServiceBroker
19	* @typedef {import("./node-catalog")} NodeCatalogClass
20	* @typedef {import("./node-catalog").NodeCatalogListOptions} NodeCatalogListOptions
21	* @typedef {import("./node-catalog").NodeCatalogListResult} NodeCatalogListResult
22	*/
23
24	/**
25	* Catalog for nodes
26	*
27	* @class NodeCatalog
28	* @implements {NodeCatalogClass}
29	*/
30	class NodeCatalog {
31	/**
32	* Creates an instance of NodeCatalog.
33	*
34	* @param {Registry} registry
35	* @param {ServiceBroker} broker
36	*
37	* @memberof NodeCatalog
38	*/
39	constructor(registry, broker) {
40	this.registry = registry;	1,350✔
41	this.broker = broker;	1,350✔
42	this.logger = registry.logger;	1,350✔
43
44	this.localNode = null;	1,350✔
45	this.nodes = new Map();	1,350✔
46
47	this.createLocalNode();	1,350✔
48	}
49
50	/**
51	* Create local node with local information
52	*
53	* @returns
54	* @memberof NodeCatalog
55	*/
56	createLocalNode() {
57	const node = new Node(this.broker.nodeID);	1,350✔
58	node.local = true;	1,350✔
59	node.ipList = getIpList();	1,350✔
60	node.instanceID = this.broker.instanceID;	1,350✔
61	node.hostname = os.hostname();	1,350✔
62	node.client = {	1,350✔
63	type: "nodejs",
64	version: this.broker.MOLECULER_VERSION,
65	langVersion: process.version
66	};
67	node.metadata = this.broker.metadata;	1,350✔
68	node.seq = 1;	1,350✔
69
70	this.add(node.id, node);	1,350✔
71
72	this.localNode = node;	1,350✔
73	return node;	1,350✔
74	}
75
76	/**
77	* Add a new node
78	*
79	* @param {String} id
80	* @param {Node} node
81	* @memberof NodeCatalog
82	*/
83	add(id, node) {
84	this.nodes.set(id, node);	1,610✔
85	}
86
87	/**
88	* Check a node exist by nodeID
89	*
90	* @param {String} id
91	* @returns
92	* @memberof NodeCatalog
93	*/
94	has(id) {
95	return this.nodes.has(id);	8✔
96	}
97
98	/**
99	* Get a node by nodeID
100	*
101	* @param {String} id
102	* @returns {Node}
103	* @memberof NodeCatalog
104	*/
105	get(id) {
106	return this.nodes.get(id);	652✔
107	}
108
109	/**
110	* Delete a node by nodeID
111	*
112	* @param {String} id
113	* @returns
114	* @memberof NodeCatalog
115	*/
116	delete(id) {
117	return this.nodes.delete(id);	×
118	}
119
120	/**
121	* Get count of all registered nodes
122	*/
123	count() {
124	return this.nodes.size;	60✔
125	}
126
127	/**
128	* Get count of online nodes
129	*/
130	onlineCount() {
131	let count = 0;	60✔
132	this.nodes.forEach(node => {	60✔
133	if (node.available) count++;	62!
134	});
135
136	return count;	60✔
137	}
138
139	/**
140	* Process incoming INFO packet payload
141	*
142	* @param {any} payload
143	* @returns {Node}
144	* @memberof NodeCatalog
145	*/
146	processNodeInfo(payload) {
147	const nodeID = payload.sender;	492✔
148	//let oldNode;
149	let node = this.get(nodeID);	492✔
150	let isNew = false;	492✔
151	let isReconnected = false;	492✔
152
153	if (!node) {	492✔
154	isNew = true;	250✔
155	node = new Node(nodeID);	250✔
156
157	this.add(nodeID, node);	250✔
158	} else if (!node.available) {	242✔
159	isReconnected = true;	10✔
160	node.lastHeartbeatTime = Math.round(process.uptime());	10✔
161	node.available = true;	10✔
162	node.offlineSince = null;	10✔
163	}
164
165	// Update instance
166	const needRegister = node.update(payload, isReconnected);	492✔
167
168	// Refresh services if 'seq' is greater or it is a reconnected node
169	if (needRegister && node.services) {	492✔
170	this.registry.registerServices(node, node.services);	484✔
171	}
172
173	// Local notifications
174	if (isNew) {	492✔
175	this.broker.broadcastLocal("$node.connected", { node, reconnected: false });	250✔
176	this.logger.info(`Node '${nodeID}' connected.`);	250✔
177	this.registry.updateMetrics();	250✔
178	} else if (isReconnected) {	242✔
179	this.broker.broadcastLocal("$node.connected", { node, reconnected: true });	10✔
180	this.logger.info(`Node '${nodeID}' reconnected.`);	10✔
181	this.registry.updateMetrics();	10✔
182	} else {
183	this.broker.broadcastLocal("$node.updated", { node });	232✔
184	this.logger.debug(`Node '${nodeID}' updated.`);	232✔
185	}
186
187	return node;	492✔
188	}
189
190	/**
191	* Disconnected a node
192	*
193	* @param {String} nodeID
194	* @param {Boolean} isUnexpected
195	* @memberof NodeCatalog
196	*/
197	disconnected(nodeID, isUnexpected) {
198	let node = this.get(nodeID);	126✔
199	if (node && node.available) {	126!
200	node.disconnected(isUnexpected);	126✔
201
202	this.registry.unregisterServicesByNode(node.id);	126✔
203
204	this.broker.broadcastLocal("$node.disconnected", { node, unexpected: !!isUnexpected });	126✔
205
206	this.broker.servicesChanged(false);	126✔
207
208	this.registry.updateMetrics();	126✔
209
210	if (isUnexpected) this.logger.warn(`Node '${node.id}' disconnected unexpectedly.`);	126✔
211	else this.logger.info(`Node '${node.id}' disconnected.`);	124✔
212
213	if (this.broker.transit) this.broker.transit.removePendingRequestByNodeID(nodeID);	126!
214	}
215	}
216
217	/**
218	* Get a node list
219	*
220	* @param {NodeCatalogListOptions} opts
221	* @returns {NodeCatalogListResult[]}
222	* @memberof NodeCatalog
223	*/
224	list({ onlyAvailable = false, withServices = false } = {}) {	5!
225	let res = [];	10✔
226	this.nodes.forEach(node => {	10✔
227	if (onlyAvailable && !node.available) return;	18✔
228
229	if (withServices) res.push(_.omit(node, ["rawInfo"]));	16✔
230	else res.push(_.omit(node, ["services", "rawInfo"]));	12✔
231	});
232
233	return res;	10✔
234	}
235
236	/**
237	* Get a copy from node list.
238	*/
239	toArray() {
240	return Array.from(this.nodes.values());	4✔
241	}
242	}
243
244	module.exports = NodeCatalog;	200✔

moleculerjs / moleculer / 29819832096

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