7062896254

Committed 01 Dec 2023 04:58PM UTC coverage: 94.16% (-0.001%) from 94.161%

Build # 7062896254

Build Type

push

github

Committed by

icebob

Commit Message

fix tests

Run Details

5314 of 5865 branches covered (0.0%)

Branch coverage included in aggregate %.

7198 of 7423 relevant lines covered (96.97%)

1498.34 hits per line

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

95.54

/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.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");	606✔
10	const os = require("os");	606✔
11	const Node = require("./node");	606✔
12	const { getIpList } = require("../utils");	606✔
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;	4,032✔
41	this.broker = broker;	4,032✔
42	this.logger = registry.logger;	4,032✔
43
44	this.localNode = null;	4,032✔
45	this.nodes = new Map();	4,032✔
46
47	this.createLocalNode();	4,032✔
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);	4,032✔
58	node.local = true;	4,032✔
59	node.ipList = getIpList();	4,032✔
60	node.instanceID = this.broker.instanceID;	4,032✔
61	node.hostname = os.hostname();	4,032✔
62	node.client = {	4,032✔
63	type: "nodejs",
64	version: this.broker.MOLECULER_VERSION,
65	langVersion: process.version
66	};
67	node.metadata = this.broker.metadata;	4,032✔
68	node.seq = 1;	4,032✔
69
70	this.add(node.id, node);	4,032✔
71
72	this.localNode = node;	4,032✔
73	return node;	4,032✔
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);	4,812✔
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);	24✔
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);	1,956✔
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;	180✔
125	}
126
127	/**
128	* Get count of online nodes
129	*/
130	onlineCount() {
131	let count = 0;	180✔
132	this.nodes.forEach(node => {	180✔
133	if (node.available) count++;	186!
134	});
135
136	return count;	180✔
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;	1,476✔
148	//let oldNode;
149	let node = this.get(nodeID);	1,476✔
150	let isNew = false;	1,476✔
151	let isReconnected = false;	1,476✔
152
153	if (!node) {	1,476✔
154	isNew = true;	750✔
155	node = new Node(nodeID);	750✔
156
157	this.add(nodeID, node);	750✔
158	} else if (!node.available) {	726✔
159	isReconnected = true;	30✔
160	node.lastHeartbeatTime = Math.round(process.uptime());	30✔
161	node.available = true;	30✔
162	node.offlineSince = null;	30✔
163	}
164
165	// Update instance
166	const needRegister = node.update(payload, isReconnected);	1,476✔
167
168	// Refresh services if 'seq' is greater or it is a reconnected node
169	if (needRegister && node.services) {	1,476✔
170	this.registry.registerServices(node, node.services);	1,452✔
171	}
172
173	// Local notifications
174	if (isNew) {	1,476✔
175	this.broker.broadcastLocal("$node.connected", { node, reconnected: false });	750✔
176	this.logger.info(`Node '${nodeID}' connected.`);	750✔
177	this.registry.updateMetrics();	750✔
178	} else if (isReconnected) {	726✔
179	this.broker.broadcastLocal("$node.connected", { node, reconnected: true });	30✔
180	this.logger.info(`Node '${nodeID}' reconnected.`);	30✔
181	this.registry.updateMetrics();	30✔
182	} else {
183	this.broker.broadcastLocal("$node.updated", { node });	696✔
184	this.logger.debug(`Node '${nodeID}' updated.`);	696✔
185	}
186
187	return node;	1,476✔
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);	378✔
199	if (node && node.available) {	378!
200	node.disconnected(isUnexpected);	378✔
201
202	this.registry.unregisterServicesByNode(node.id);	378✔
203
204	this.broker.broadcastLocal("$node.disconnected", { node, unexpected: !!isUnexpected });	378✔
205
206	this.registry.updateMetrics();	378✔
207
208	if (isUnexpected) this.logger.warn(`Node '${node.id}' disconnected unexpectedly.`);	378✔
209	else this.logger.info(`Node '${node.id}' disconnected.`);	372✔
210
211	if (this.broker.transit) this.broker.transit.removePendingRequestByNodeID(nodeID);	378!
212	}
213	}
214
215	/**
216	* Get a node list
217	*
218	* @param {NodeCatalogListOptions} opts
219	* @returns {NodeCatalogListResult[]}
220	* @memberof NodeCatalog
221	*/
222	list({ onlyAvailable = false, withServices = false } = {}) {	15!
223	let res = [];	30✔
224	this.nodes.forEach(node => {	30✔
225	if (onlyAvailable && !node.available) return;	54✔
226
227	if (withServices) res.push(_.omit(node, ["rawInfo"]));	48✔
228	else res.push(_.omit(node, ["services", "rawInfo"]));	36✔
229	});
230
231	return res;	30✔
232	}
233
234	/**
235	* Get a copy from node list.
236	*/
237	toArray() {
238	return Array.from(this.nodes.values());	12✔
239	}
240	}
241
242	module.exports = NodeCatalog;	606✔

moleculerjs / moleculer / 7062896254

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