7065643145

Committed 01 Dec 2023 10:01PM UTC coverage: 94.161% (+0.001%) from 94.16%

Build # 7065643145

Build Type

push

github

Committed by

icebob

Commit Message

fix tests & types

Run Details

5315 of 5866 branches covered (0.0%)

Branch coverage included in aggregate %.

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

28 existing lines in 7 files now uncovered.

7200 of 7425 relevant lines covered (96.97%)

1498.02 hits per line

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

97.44

/src/registry/discoverers/base.js

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

/* eslint-disable no-unused-vars */

"use strict";

const _ = require("lodash");

/**
 * Import types
 *
 * @typedef {import("./base")} BaseDiscovererClass
 * @typedef {import("../registry")} ServiceRegistry
 */

/**
 * Abstract Discoverer class
 *
 * @class BaseDiscoverer
 * @implements {BaseDiscovererClass}
 */
class BaseDiscoverer {
        /**
         * Creates an instance of Discoverer.
         *
         * @memberof BaseDiscoverer
         */
        constructor(opts) {
                this.Promise = Promise; // while `init` is not called

                this.opts = _.defaultsDeep({}, opts, {
                        heartbeatInterval: null,
                        heartbeatTimeout: null,

                        disableHeartbeatChecks: false,
                        disableOfflineNodeRemoving: false,
                        cleanOfflineNodesTimeout: 10 * 60 // 10 minutes
                });

                // Timer variables
                this.heartbeatTimer = null;
                this.checkNodesTimer = null;
                this.offlineTimer = null;

                // Pointer for the local `Node` instance
                this.localNode = null;
        }

        /**
         * Initialize Discoverer
         *
         * @param {ServiceRegistry} registry
         */
        init(registry) {
                this.registry = registry;
                this.broker = registry.broker;
                this.Promise = this.broker.Promise;

                if (this.broker) {
                        this.logger = this.broker.getLogger("Discovery");
                        this.transit = this.broker.transit;

                        // Get HB time settings from broker options. Backward compatibility
                        if (this.opts.heartbeatInterval == null)
                                this.opts.heartbeatInterval = this.broker.options.heartbeatInterval;
                        if (this.opts.heartbeatTimeout == null)
                                this.opts.heartbeatTimeout = this.broker.options.heartbeatTimeout;
                }

                if (this.transit) {
                        this.broker.localBus.on("$transporter.connected", () => this.startHeartbeatTimers());
                        this.broker.localBus.on("$transporter.disconnected", () => this.stopHeartbeatTimers());
                }

                this.localNode = this.registry.nodes.localNode;

                this.registerMoleculerMetrics();
        }

        /**
         * Stop discoverer clients.
         */
        stop() {
                this.stopHeartbeatTimers();
                return this.Promise.resolve();
        }

        /**
         * Register Moleculer Transit Core metrics.
         */
        registerMoleculerMetrics() {
                // Not implemented
        }

        /**
         * Start heartbeat timers
         */
        startHeartbeatTimers() {
                this.stopHeartbeatTimers();

                if (this.opts.heartbeatInterval > 0) {
                        // HB timer
                        const time =
                                this.opts.heartbeatInterval * 1000 + (Math.round(Math.random() * 1000) - 500); // random +/- 500ms
                        this.heartbeatTimer = setInterval(() => this.beat(), time);
                        this.heartbeatTimer.unref();

                        // Check expired heartbeats of remote nodes timer
                        this.checkNodesTimer = setInterval(
                                () => this.checkRemoteNodes(),
                                this.opts.heartbeatTimeout * 1000
                        );
                        this.checkNodesTimer.unref();

                        // Clean offline nodes timer
                        this.offlineTimer = setInterval(() => this.checkOfflineNodes(), 60 * 1000); // 1 min
                        this.offlineTimer.unref();
                }
        }

        /**
         * Stop heartbeat timers
         */
        stopHeartbeatTimers() {
                if (this.heartbeatTimer) {
                        clearInterval(this.heartbeatTimer);
                        this.heartbeatTimer = null;
                }

                if (this.checkNodesTimer) {
                        clearInterval(this.checkNodesTimer);
                        this.checkNodesTimer = null;
                }

                if (this.offlineTimer) {
                        clearInterval(this.offlineTimer);
                        this.offlineTimer = null;
                }
        }

        /**
         * Disable built-in Heartbeat logic. Used by TCP transporter
         */
        disableHeartbeat() {
                this.opts.heartbeatInterval = 0;
                this.stopHeartbeatTimers();
        }

        /**
         * Heartbeat method.
         */
        beat() {
                // Update the local CPU usage before sending heartbeat.
                return this.localNode
                        .updateLocalInfo(this.broker.getCpuUsage)
                        .then(() => this.sendHeartbeat());
        }

        /**
         * Check all registered remote nodes are available.
         */
        checkRemoteNodes() {
                if (this.opts.disableHeartbeatChecks) return;

                const now = Math.round(process.uptime());
                this.registry.nodes.toArray().forEach(node => {
                        if (node.local || !node.available) return;
                        if (!node.lastHeartbeatTime) {
                                // Not received the first heartbeat yet
                                node.lastHeartbeatTime = now;
                                return;
                        }

                        if (now - node.lastHeartbeatTime > this.opts.heartbeatTimeout) {
                                this.logger.warn(`Heartbeat is not received from '${node.id}' node.`);
                                this.registry.nodes.disconnected(node.id, true);
                        }
                });
        }

        /**
         * Check offline nodes. Remove which is older than 10 minutes.
         */
        checkOfflineNodes() {
                if (this.opts.disableOfflineNodeRemoving || !this.opts.cleanOfflineNodesTimeout) return;

                const now = Math.round(process.uptime());
                this.registry.nodes.toArray().forEach(node => {
                        if (node.local || node.available) return;
                        if (!node.lastHeartbeatTime) {
                                // Not received the first
                                node.lastHeartbeatTime = now;
                                return;
                        }

                        if (now - node.lastHeartbeatTime > this.opts.cleanOfflineNodesTimeout) {
                                this.logger.warn(
                                        `Removing offline '${node.id}' node from registry because it hasn't submitted heartbeat signal for 10 minutes.`
                                );
                                this.registry.nodes.delete(node.id);
                        }
                });
        }

        /**
         * Heartbeat received from a remote node.
         *
         * @param {String} nodeID
         * @param {Object} payload
         */
        heartbeatReceived(nodeID, payload) {
                const node = this.registry.nodes.get(nodeID);
                if (node) {
                        if (!node.available) {
                                // Reconnected node. Request a fresh INFO
                                this.discoverNode(nodeID);
                        } else {
                                if (payload.seq != null && node.seq !== payload.seq) {
                                        // Some services changed on the remote node. Request a new INFO
                                        this.discoverNode(nodeID);
                                } else if (
                                        payload.instanceID != null &&
                                        !node.instanceID.startsWith(payload.instanceID)
                                ) {
                                        // The node has been restarted. Request a new INFO
                                        this.discoverNode(nodeID);
                                } else {
                                        node.heartbeat(payload);
                                }
                        }
                } else {
                        // Unknow node. Request an INFO
                        this.discoverNode(nodeID);
                }
        }

        /**
         * Received an INFO from a remote node.
         *
         * @param {String} nodeID
         * @param {Object} payload
         */
        processRemoteNodeInfo(nodeID, payload) {
                return this.broker.registry.processNodeInfo(payload);
        }

        /**
         * Sending a local heartbeat to remote nodes.
         */
        sendHeartbeat() {
                if (!this.transit) return this.Promise.resolve();
                return this.transit.sendHeartbeat(this.localNode);
        }

        /**
         * Discover a new or old node by nodeID
         *
         * @param {String} nodeID
         */
        discoverNode(nodeID) {
                /* istanbul ignore next */
                throw new Error("Not implemented");
        }

        /**
         * Discover all nodes (after connected)
         */
        discoverAllNodes() {
                /* istanbul ignore next */
                throw new Error("Not implemented");
        }

        /**
         * Local service registry has been changed. We should notify remote nodes.
         *
         * @param {String=} nodeID
         */
        sendLocalNodeInfo(nodeID) {
                /* istanbul ignore next */
                throw new Error("Not implemented");
        }

        /**
         * Called when the local node disconnected.
         * You can clean it from the remote registry.
         */
        localNodeDisconnected() {
                if (!this.transit) return this.Promise.resolve();
                return this.transit.sendDisconnectPacket();
        }

        /**
         * Called when a remote node disconnected (received DISCONNECT packet)
         * You can clean it from local registry.
         *
         * @param {String} nodeID
         * @param {Boolean} isUnexpected
         */
        remoteNodeDisconnected(nodeID, isUnexpected) {
                return this.registry.nodes.disconnected(nodeID, isUnexpected);
        }
}

module.exports = BaseDiscoverer;

1	/*
2	* moleculer
3	* Copyright (c) 2023 MoleculerJS (https://github.com/moleculerjs/moleculer)
4	* MIT Licensed
5	*/
6
7	/* eslint-disable no-unused-vars */
8
9	"use strict";
10
11	const _ = require("lodash");	612✔
12
13	/**
14	* Import types
15	*
16	* @typedef {import("./base")} BaseDiscovererClass
17	* @typedef {import("../registry")} ServiceRegistry
18	*/
19
20	/**
21	* Abstract Discoverer class
22	*
23	* @class BaseDiscoverer
24	* @implements {BaseDiscovererClass}
25	*/
26	class BaseDiscoverer {
27	/**
28	* Creates an instance of Discoverer.
29	*
30	* @memberof BaseDiscoverer
31	*/
32	constructor(opts) {
33	this.Promise = Promise; // while `init` is not called	4,398✔
34
35	this.opts = _.defaultsDeep({}, opts, {	4,398✔
36	heartbeatInterval: null,
37	heartbeatTimeout: null,
38
39	disableHeartbeatChecks: false,
40	disableOfflineNodeRemoving: false,
41	cleanOfflineNodesTimeout: 10 * 60 // 10 minutes
42	});
43
44	// Timer variables
45	this.heartbeatTimer = null;	4,398✔
46	this.checkNodesTimer = null;	4,398✔
47	this.offlineTimer = null;	4,398✔
48
49	// Pointer for the local `Node` instance
50	this.localNode = null;	4,398✔
51	}
52
53	/**
54	* Initialize Discoverer
55	*
56	* @param {ServiceRegistry} registry
57	*/
58	init(registry) {
59	this.registry = registry;	4,278✔
60	this.broker = registry.broker;	4,278✔
61	this.Promise = this.broker.Promise;	4,278✔
62
63	if (this.broker) {	4,278!
64	this.logger = this.broker.getLogger("Discovery");	4,278✔
65	this.transit = this.broker.transit;	4,278✔
66
67	// Get HB time settings from broker options. Backward compatibility
68	if (this.opts.heartbeatInterval == null)	4,278✔
69	this.opts.heartbeatInterval = this.broker.options.heartbeatInterval;	4,242✔
70	if (this.opts.heartbeatTimeout == null)	4,278✔
71	this.opts.heartbeatTimeout = this.broker.options.heartbeatTimeout;	4,236✔
72	}
73
74	if (this.transit) {	4,278✔
75	this.broker.localBus.on("$transporter.connected", () => this.startHeartbeatTimers());	936✔
76	this.broker.localBus.on("$transporter.disconnected", () => this.stopHeartbeatTimers());	936✔
77	}
78
79	this.localNode = this.registry.nodes.localNode;	4,278✔
80
81	this.registerMoleculerMetrics();	4,278✔
82	}
83
84	/**
85	* Stop discoverer clients.
86	*/
87	stop() {
88	this.stopHeartbeatTimers();	1,086✔
89	return this.Promise.resolve();	1,086✔
90	}
91
92	/**
93	* Register Moleculer Transit Core metrics.
94	*/
95	registerMoleculerMetrics() {
96	// Not implemented
97	}
98
99	/**
100	* Start heartbeat timers
101	*/
102	startHeartbeatTimers() {
103	this.stopHeartbeatTimers();	348✔
104
105	if (this.opts.heartbeatInterval > 0) {	348!
106	// HB timer
107	const time =
108	this.opts.heartbeatInterval * 1000 + (Math.round(Math.random() * 1000) - 500); // random +/- 500ms	348✔
109	this.heartbeatTimer = setInterval(() => this.beat(), time);	348✔
110	this.heartbeatTimer.unref();	348✔
111
112	// Check expired heartbeats of remote nodes timer
113	this.checkNodesTimer = setInterval(	348✔
114	() => this.checkRemoteNodes(),	12✔
115	this.opts.heartbeatTimeout * 1000
116	);
117	this.checkNodesTimer.unref();	348✔
118
119	// Clean offline nodes timer
120	this.offlineTimer = setInterval(() => this.checkOfflineNodes(), 60 * 1000); // 1 min	348✔
121	this.offlineTimer.unref();	348✔
122	}
123	}
124
125	/**
126	* Stop heartbeat timers
127	*/
128	stopHeartbeatTimers() {
129	if (this.heartbeatTimer) {	2,052✔
130	clearInterval(this.heartbeatTimer);	312✔
131	this.heartbeatTimer = null;	312✔
132	}
133
134	if (this.checkNodesTimer) {	2,052✔
135	clearInterval(this.checkNodesTimer);	312✔
136	this.checkNodesTimer = null;	312✔
137	}
138
139	if (this.offlineTimer) {	2,052✔
140	clearInterval(this.offlineTimer);	312✔
141	this.offlineTimer = null;	312✔
142	}
143	}
144
145	/**
146	* Disable built-in Heartbeat logic. Used by TCP transporter
147	*/
148	disableHeartbeat() {
149	this.opts.heartbeatInterval = 0;	306✔
150	this.stopHeartbeatTimers();	306✔
151	}
152
153	/**
154	* Heartbeat method.
155	*/
156	beat() {
157	// Update the local CPU usage before sending heartbeat.
158	return this.localNode	18✔
159	.updateLocalInfo(this.broker.getCpuUsage)
160	.then(() => this.sendHeartbeat());	18✔
161	}
162
163	/**
164	* Check all registered remote nodes are available.
165	*/
166	checkRemoteNodes() {
167	if (this.opts.disableHeartbeatChecks) return;	36✔
168
169	const now = Math.round(process.uptime());	30✔
170	this.registry.nodes.toArray().forEach(node => {	30✔
171	if (node.local \|\| !node.available) return;	30✔
172	if (!node.lastHeartbeatTime) {	18✔
173	// Not received the first heartbeat yet
174	node.lastHeartbeatTime = now;	6✔
175	return;	6✔
176	}
177
178	if (now - node.lastHeartbeatTime > this.opts.heartbeatTimeout) {	12✔
179	this.logger.warn(`Heartbeat is not received from '${node.id}' node.`);	6✔
180	this.registry.nodes.disconnected(node.id, true);	6✔
181	}
182	});
183	}
184
185	/**
186	* Check offline nodes. Remove which is older than 10 minutes.
187	*/
188	checkOfflineNodes() {
189	if (this.opts.disableOfflineNodeRemoving \|\| !this.opts.cleanOfflineNodesTimeout) return;	36✔
190
191	const now = Math.round(process.uptime());	30✔
192	this.registry.nodes.toArray().forEach(node => {	30✔
193	if (node.local \|\| node.available) return;	30✔
194	if (!node.lastHeartbeatTime) {	18✔
195	// Not received the first
196	node.lastHeartbeatTime = now;	6✔
197	return;	6✔
198	}
199
200	if (now - node.lastHeartbeatTime > this.opts.cleanOfflineNodesTimeout) {	12✔
201	this.logger.warn(	6✔
202	`Removing offline '${node.id}' node from registry because it hasn't submitted heartbeat signal for 10 minutes.`
203	);
204	this.registry.nodes.delete(node.id);	6✔
205	}
206	});
207	}
208
209	/**
210	* Heartbeat received from a remote node.
211	*
212	* @param {String} nodeID
213	* @param {Object} payload
214	*/
215	heartbeatReceived(nodeID, payload) {
216	const node = this.registry.nodes.get(nodeID);	66✔
217	if (node) {	66!
218	if (!node.available) {	66✔
219	// Reconnected node. Request a fresh INFO
220	this.discoverNode(nodeID);	12✔
221	} else {
222	if (payload.seq != null && node.seq !== payload.seq) {	54✔
223	// Some services changed on the remote node. Request a new INFO
224	this.discoverNode(nodeID);	6✔
225	} else if (	48✔
226	payload.instanceID != null &&	30✔
227	!node.instanceID.startsWith(payload.instanceID)
228	) {
229	// The node has been restarted. Request a new INFO
230	this.discoverNode(nodeID);	6✔
231	} else {
232	node.heartbeat(payload);	42✔
233	}
234	}
235	} else {
236	// Unknow node. Request an INFO
UNCOV 237	this.discoverNode(nodeID);	×
238	}
239	}
240
241	/**
242	* Received an INFO from a remote node.
243	*
244	* @param {String} nodeID
245	* @param {Object} payload
246	*/
247	processRemoteNodeInfo(nodeID, payload) {
248	return this.broker.registry.processNodeInfo(payload);	1,440✔
249	}
250
251	/**
252	* Sending a local heartbeat to remote nodes.
253	*/
254	sendHeartbeat() {
255	if (!this.transit) return this.Promise.resolve();	24✔
256	return this.transit.sendHeartbeat(this.localNode);	18✔
257	}
258
259	/**
260	* Discover a new or old node by nodeID
261	*
262	* @param {String} nodeID
263	*/
264	discoverNode(nodeID) {
265	/* istanbul ignore next */
266	throw new Error("Not implemented");
267	}
268
269	/**
270	* Discover all nodes (after connected)
271	*/
272	discoverAllNodes() {
273	/* istanbul ignore next */
274	throw new Error("Not implemented");
275	}
276
277	/**
278	* Local service registry has been changed. We should notify remote nodes.
279	*
280	* @param {String=} nodeID
281	*/
282	sendLocalNodeInfo(nodeID) {
283	/* istanbul ignore next */
284	throw new Error("Not implemented");
285	}
286
287	/**
288	* Called when the local node disconnected.
289	* You can clean it from the remote registry.
290	*/
291	localNodeDisconnected() {
292	if (!this.transit) return this.Promise.resolve();	324✔
293	return this.transit.sendDisconnectPacket();	306✔
294	}
295
296	/**
297	* Called when a remote node disconnected (received DISCONNECT packet)
298	* You can clean it from local registry.
299	*
300	* @param {String} nodeID
301	* @param {Boolean} isUnexpected
302	*/
303	remoteNodeDisconnected(nodeID, isUnexpected) {
304	return this.registry.nodes.disconnected(nodeID, isUnexpected);	366✔
305	}
306	}
307
308	module.exports = BaseDiscoverer;	612✔

moleculerjs / moleculer / 7065643145

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