7157014227

Committed 10 Dec 2023 09:51AM UTC coverage: 94.12% (-0.05%) from 94.173%

Build # 7157014227

Build Type

push

github

Committed by

icebob

Commit Message

update redis creating

Run Details

5285 of 5836 branches covered (0.0%)

Branch coverage included in aggregate %.

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

78 existing lines in 14 files now uncovered.

7169 of 7396 relevant lines covered (96.93%)

1000.72 hits per line

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

90.27

/src/registry/discoverers/etcd3.js

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

"use strict";

const _ = require("lodash");
const kleur = require("kleur");
const BaseDiscoverer = require("./base");
const { METRIC } = require("../../metrics");
const Serializers = require("../../serializers");
const { removeFromArray, randomInt } = require("../../utils");
const P = require("../../packets");
const C = require("../../constants");

let ETCD3;

/**
 * Import types
 *
 * @typedef {import("./etcd3")} Etcd3DiscovererClass
 * @typedef {import("./etcd3").Etcd3DiscovererOptions} Etcd3DiscovererOptions
 * @typedef {import("../node")} Node
 */

/**
 * etcd3-based Discoverer class
 *
 * @class Etcd3Discoverer
 * @implements {Etcd3DiscovererClass}
 */
class Etcd3Discoverer extends BaseDiscoverer {
        /**
         * Creates an instance of Discoverer.
         *
         * TODO:
         *         - the etcd3 lib has no reconnection logic
         *
         * @param {Etcd3DiscovererOptions|string} opts
         * @memberof Etcd3Discoverer
         */
        constructor(opts) {
                if (typeof opts === "string") opts = { etcd: { hosts: opts.replace(/etcd3:\/\//g, "") } };

                super(opts);

                this.opts = _.defaultsDeep(this.opts, {
                        etcd: undefined,
                        serializer: "JSON",
                        fullCheck: 10 // Disable with `0` or `null`
                });

                // Loop counter for full checks. Starts from a random value for better distribution
                this.idx = this.opts.fullCheck > 1 ? randomInt(this.opts.fullCheck - 1) : 0;

                // Etcd client instance
                this.client = null;

                // Last sequence numbers
                this.lastInfoSeq = 0;
                this.lastBeatSeq = 0;

                // Leases
                this.leaseBeat = null;
                this.leaseInfo = null;
        }

        /**
         * Initialize Discoverer
         *
         * @param {any} registry
         *
         * @memberof Etcd3Discoverer
         */
        init(registry) {
                super.init(registry);

                try {
                        ETCD3 = require("etcd3");
                } catch (err) {
                        /* istanbul ignore next */
                        this.broker.fatal(
                                "The 'etcd3' package is missing. Please install it with 'npm install etcd3 --save' command.",
                                err,
                                true
                        );
                }

                this.logger.warn(
                        kleur
                                .yellow()
                                .bold("Etcd3 Discoverer is an EXPERIMENTAL module. Do NOT use it in production!")
                );

                this.instanceHash = this.broker.instanceID.substring(0, 8);

                this.PREFIX = `moleculer${
                        this.broker.namespace ? "-" + this.broker.namespace : ""
                }/discovery`;
                this.BEAT_KEY = `${this.PREFIX}/beats/${this.broker.nodeID}/${this.instanceHash}`;
                this.INFO_KEY = `${this.PREFIX}/info/${this.broker.nodeID}`;

                this.client = new ETCD3.Etcd3(this.opts.etcd);

                // create an instance of serializer (default to JSON)
                this.serializer = Serializers.resolve(this.opts.serializer);
                this.serializer.init(this.broker);

                this.logger.debug("Etcd3 Discoverer created. Prefix:", this.PREFIX);
        }

        /**
         * Stop discoverer clients.
         */
        stop() {
                return super.stop().then(() => {
                        if (this.client) return this.client.close();
                });
        }

        /**
         * Register Moleculer Transit Core metrics.
         */
        registerMoleculerMetrics() {
                this.broker.metrics.register({
                        name: METRIC.MOLECULER_DISCOVERER_ETCD_COLLECT_TOTAL,
                        type: METRIC.TYPE_COUNTER,
                        rate: true,
                        description: "Number of Service Registry fetching from etcd"
                });
                this.broker.metrics.register({
                        name: METRIC.MOLECULER_DISCOVERER_ETCD_COLLECT_TIME,
                        type: METRIC.TYPE_HISTOGRAM,
                        quantiles: true,
                        unit: METRIC.UNIT_MILLISECONDS,
                        description: "Time of Service Registry fetching from etcd"
                });
        }

        /**
         * Sending a local heartbeat to etcd.
         */
        sendHeartbeat() {
                const timeEnd = this.broker.metrics.timer(METRIC.MOLECULER_DISCOVERER_ETCD_COLLECT_TIME);
                const data = {
                        sender: this.broker.nodeID,
                        ver: this.broker.PROTOCOL_VERSION,

                        //timestamp: Date.now(),
                        cpu: this.localNode.cpu,
                        seq: this.localNode.seq,
                        instanceID: this.broker.instanceID
                };

                const seq = this.localNode.seq;
                const key = this.BEAT_KEY + "/" + seq;
                let leaseBeat = this.leaseBeat;

                return this.Promise.resolve()
                        .then(() => {
                                if (leaseBeat) {
                                        if (seq != this.lastBeatSeq) {
                                                // If seq changed, revoke the current lease
                                                const p = leaseBeat.revoke();
                                                leaseBeat = null;
                                                return p;
                                        }
                                }
                        })
                        .then(() => {
                                if (!leaseBeat) {
                                        // Create a new for lease
                                        leaseBeat = this.client.lease(this.opts.heartbeatTimeout);

                                        //Handle lease-lost event. Release lease when lost. Next heartbeat will request a new lease
                                        leaseBeat.on("lost", err => {
                                                this.logger.warn(
                                                        "Lost heartbeat lease. Dropping lease and retrying heartbeat. Error:",
                                                        err.message
                                                );
                                                leaseBeat.release();
                                                this.leaseBeat = null;
                                                // if broker is connected, send heartbeat immediately. Otherwise it is sent on reconnect.
                                                if (this.broker.transit.connected) {
                                                        this.sendHeartbeat();
                                                }
                                        });

                                        return leaseBeat
                                                .grant() // Waiting for the lease creation on the server
                                                .then(() => (this.leaseBeat = leaseBeat));
                                }
                        })
                        .then(() =>
                                this.leaseBeat.put(key).value(this.serializer.serialize(data, P.PACKET_HEARTBEAT))
                        )
                        .then(() => (this.lastBeatSeq = seq))
                        .then(() => this.collectOnlineNodes())
                        .catch(err => {
                                this.logger.error("Error occurred while collect etcd keys.", err);

                                this.broker.broadcastLocal("$discoverer.error", {
                                        error: err,
                                        module: "discoverer",
                                        type: C.FAILED_COLLECT_KEYS
                                });
                        })
                        .then(() => {
                                timeEnd();
                                this.broker.metrics.increment(METRIC.MOLECULER_DISCOVERER_ETCD_COLLECT_TOTAL);
                        });
        }

        /**
         * Collect online nodes from etcd server.
         */
        collectOnlineNodes() {
                // Get the current node list so that we can check the disconnected nodes.
                const prevNodes = this.registry.nodes
                        .list({ onlyAvailable: true, withServices: false })
                        .map(node => node.id)
                        .filter(nodeID => nodeID !== this.broker.nodeID);

                // Collect the online node keys.
                return this.Promise.resolve()
                        .then(() => {
                                if (this.opts.fullCheck && ++this.idx % this.opts.fullCheck === 0) {
                                        // Full check
                                        //this.logger.debug("Full check", this.idx);
                                        this.idx = 0;

                                        return this.client
                                                .getAll()
                                                .prefix(`${this.PREFIX}/beats/`)
                                                .buffers()
                                                .then(result =>
                                                        Object.values(result).map(raw => {
                                                                try {
                                                                        return this.serializer.deserialize(raw, P.PACKET_INFO);
                                                                } catch (err) {
                                                                        this.logger.warn("Unable to parse HEARTBEAT packet", err, raw);
                                                                }
                                                        })
                                                );
                                } else {
                                        //this.logger.debug("Lazy check", this.idx);
                                        // Lazy check
                                        return this.client
                                                .getAll()
                                                .prefix(`${this.PREFIX}/beats/`)
                                                .keys()
                                                .then(keys =>
                                                        keys.map(key => {
                                                                const p = key.substring(`${this.PREFIX}/beats/`.length).split("/");
                                                                return {
                                                                        key,
                                                                        sender: p[0],
                                                                        instanceID: p[1],
                                                                        seq: Number(p[2])
                                                                };
                                                        })
                                                );
                                }
                        })

                        .then(packets => {
                                _.compact(packets).map(packet => {
                                        if (packet.sender == this.broker.nodeID) return;

                                        removeFromArray(prevNodes, packet.sender);
                                        this.heartbeatReceived(packet.sender, packet);
                                });
                        })

                        .then(() => {
                                if (prevNodes.length > 0) {
                                        // Disconnected nodes
                                        prevNodes.map(nodeID => {
                                                this.logger.info(
                                                        `The node '${nodeID}' is not available. Removing from registry...`
                                                );
                                                this.remoteNodeDisconnected(nodeID, true);
                                        });
                                }
                        });
        }

        /**
         * Discover a new or old node.
         *
         * @param {String} nodeID
         * @returns {Promise<Node | void>}
         */
        discoverNode(nodeID) {
                return this.client
                        .get(`${this.PREFIX}/info/${nodeID}`)
                        .buffer()
                        .then(res => {
                                if (!res) {
                                        this.logger.warn(`No INFO for '${nodeID}' node in registry.`);
                                        return;
                                }
                                try {
                                        const info = this.serializer.deserialize(res, P.PACKET_INFO);
                                        return this.processRemoteNodeInfo(nodeID, info);
                                } catch (err) {
                                        this.logger.warn("Unable to parse INFO packet", err, res);
                                }
                        });
        }

        /**
         * Discover all nodes (after connected)
         * @returns {Promise<Node[] | void>}
         */
        discoverAllNodes() {
                return this.collectOnlineNodes();
        }

        /**
         * Local service registry has been changed. We should notify remote nodes.
         * @param {String=} nodeID
         * @returns {Promise<void>}
         */
        sendLocalNodeInfo(nodeID) {
                const info = this.broker.getLocalNodeInfo();

                const payload = Object.assign(
                        {
                                ver: this.broker.PROTOCOL_VERSION,
                                sender: this.broker.nodeID
                        },
                        info
                );

                const key = this.INFO_KEY;
                const seq = this.localNode.seq;
                let leaseInfo = this.leaseInfo;

                const p =
                        !nodeID && this.broker.options.disableBalancer
                                ? this.transit.tx.makeBalancedSubscriptions()
                                : this.Promise.resolve();
                return p
                        .then(() => {
                                if (leaseInfo) {
                                        if (seq != this.lastInfoSeq) {
                                                const p = leaseInfo.revoke();
                                                leaseInfo = null;
                                                return p;
                                        }
                                }
                        })
                        .then(() => {
                                if (!leaseInfo) {
                                        leaseInfo = this.client.lease(60);

                                        //Handle lease-lost event. Release lease when lost. Next heartbeat will request a new lease
                                        leaseInfo.on("lost", err => {
                                                this.logger.warn(
                                                        "Lost info lease. Dropping lease and retrying info-send. Error:",
                                                        err.message
                                                );
                                                leaseInfo.release();
                                                this.leaseInfo = null;
                                                // if broker is connected, send local node info immediately. Otherwise it is sent on reconnect.
                                                if (this.broker.transit.connected) {
                                                        this.sendLocalNodeInfo(nodeID);
                                                }
                                        });

                                        return leaseInfo
                                                .grant() // Waiting for the lease creation on the server
                                                .then(() => (this.leaseInfo = leaseInfo));
                                }
                        })
                        .then(() => leaseInfo.put(key).value(this.serializer.serialize(payload, P.PACKET_INFO)))
                        .then(() => {
                                this.lastInfoSeq = seq;

                                // Sending a new heartbeat because it contains the `seq`
                                if (!nodeID) return this.beat();
                        })
                        .catch(err => {
                                this.logger.error("Unable to send INFO to etcd server", err);

                                this.broker.broadcastLocal("$discoverer.error", {
                                        error: err,
                                        module: "discoverer",
                                        type: C.FAILED_SEND_INFO
                                });
                        });
        }

        /**
         * Unregister local node after disconnecting.
         */
        localNodeDisconnected() {
                return this.Promise.resolve()
                        .then(() => super.localNodeDisconnected())
                        .then(() => this.logger.debug("Remove local node from registry..."))
                        .then(() => this.client.delete().key(this.INFO_KEY))
                        .then(() => this.client.delete().key(this.BEAT_KEY))
                        .then(() => {
                                if (this.leaseBeat) return this.leaseBeat.revoke();
                        })
                        .then(() => {
                                if (this.leaseInfo) return this.leaseInfo.revoke();
                        });
        }
}

module.exports = Etcd3Discoverer;

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");	404✔
10	const kleur = require("kleur");	404✔
11	const BaseDiscoverer = require("./base");	404✔
12	const { METRIC } = require("../../metrics");	404✔
13	const Serializers = require("../../serializers");	404✔
14	const { removeFromArray, randomInt } = require("../../utils");	404✔
15	const P = require("../../packets");	404✔
16	const C = require("../../constants");	404✔
17
18	let ETCD3;
19
20	/**
21	* Import types
22	*
23	* @typedef {import("./etcd3")} Etcd3DiscovererClass
24	* @typedef {import("./etcd3").Etcd3DiscovererOptions} Etcd3DiscovererOptions
25	* @typedef {import("../node")} Node
26	*/
27
28	/**
29	* etcd3-based Discoverer class
30	*
31	* @class Etcd3Discoverer
32	* @implements {Etcd3DiscovererClass}
33	*/
34	class Etcd3Discoverer extends BaseDiscoverer {
35	/**
36	* Creates an instance of Discoverer.
37	*
38	* TODO:
39	* - the etcd3 lib has no reconnection logic
40	*
41	* @param {Etcd3DiscovererOptions\|string} opts
42	* @memberof Etcd3Discoverer
43	*/
44	constructor(opts) {
45	if (typeof opts === "string") opts = { etcd: { hosts: opts.replace(/etcd3:\/\//g, "") } };	68✔
46
47	super(opts);	68✔
48
49	this.opts = _.defaultsDeep(this.opts, {	68✔
50	etcd: undefined,
51	serializer: "JSON",
52	fullCheck: 10 // Disable with `0` or `null`
53	});
54
55	// Loop counter for full checks. Starts from a random value for better distribution
56	this.idx = this.opts.fullCheck > 1 ? randomInt(this.opts.fullCheck - 1) : 0;	68✔
57
58	// Etcd client instance
59	this.client = null;	68✔
60
61	// Last sequence numbers
62	this.lastInfoSeq = 0;	68✔
63	this.lastBeatSeq = 0;	68✔
64
65	// Leases
66	this.leaseBeat = null;	68✔
67	this.leaseInfo = null;	68✔
68	}
69
70	/**
71	* Initialize Discoverer
72	*
73	* @param {any} registry
74	*
75	* @memberof Etcd3Discoverer
76	*/
77	init(registry) {
78	super.init(registry);	40✔
79
80	try {	40✔
81	ETCD3 = require("etcd3");	40✔
82	} catch (err) {
83	/* istanbul ignore next */
84	this.broker.fatal(
85	"The 'etcd3' package is missing. Please install it with 'npm install etcd3 --save' command.",
86	err,
87	true
88	);
89	}
90
91	this.logger.warn(	40✔
92	kleur
93	.yellow()
94	.bold("Etcd3 Discoverer is an EXPERIMENTAL module. Do NOT use it in production!")
95	);
96
97	this.instanceHash = this.broker.instanceID.substring(0, 8);	40✔
98
99	this.PREFIX = `moleculer${	40✔
100	this.broker.namespace ? "-" + this.broker.namespace : ""	20✔
101	}/discovery`;
102	this.BEAT_KEY = `${this.PREFIX}/beats/${this.broker.nodeID}/${this.instanceHash}`;	40✔
103	this.INFO_KEY = `${this.PREFIX}/info/${this.broker.nodeID}`;	40✔
104
105	this.client = new ETCD3.Etcd3(this.opts.etcd);	40✔
106
107	// create an instance of serializer (default to JSON)
108	this.serializer = Serializers.resolve(this.opts.serializer);	40✔
109	this.serializer.init(this.broker);	40✔
110
111	this.logger.debug("Etcd3 Discoverer created. Prefix:", this.PREFIX);	40✔
112	}
113
114	/**
115	* Stop discoverer clients.
116	*/
117	stop() {
118	return super.stop().then(() => {	32✔
119	if (this.client) return this.client.close();	32✔
120	});
121	}
122
123	/**
124	* Register Moleculer Transit Core metrics.
125	*/
126	registerMoleculerMetrics() {
127	this.broker.metrics.register({	44✔
128	name: METRIC.MOLECULER_DISCOVERER_ETCD_COLLECT_TOTAL,
129	type: METRIC.TYPE_COUNTER,
130	rate: true,
131	description: "Number of Service Registry fetching from etcd"
132	});
133	this.broker.metrics.register({	44✔
134	name: METRIC.MOLECULER_DISCOVERER_ETCD_COLLECT_TIME,
135	type: METRIC.TYPE_HISTOGRAM,
136	quantiles: true,
137	unit: METRIC.UNIT_MILLISECONDS,
138	description: "Time of Service Registry fetching from etcd"
139	});
140	}
141
142	/**
143	* Sending a local heartbeat to etcd.
144	*/
145	sendHeartbeat() {
146	const timeEnd = this.broker.metrics.timer(METRIC.MOLECULER_DISCOVERER_ETCD_COLLECT_TIME);	16✔
147	const data = {	16✔
148	sender: this.broker.nodeID,
149	ver: this.broker.PROTOCOL_VERSION,
150
151	//timestamp: Date.now(),
152	cpu: this.localNode.cpu,
153	seq: this.localNode.seq,
154	instanceID: this.broker.instanceID
155	};
156
157	const seq = this.localNode.seq;	16✔
158	const key = this.BEAT_KEY + "/" + seq;	16✔
159	let leaseBeat = this.leaseBeat;	16✔
160
161	return this.Promise.resolve()	16✔
162	.then(() => {
163	if (leaseBeat) {	16✔
164	if (seq != this.lastBeatSeq) {	12✔
165	// If seq changed, revoke the current lease
166	const p = leaseBeat.revoke();	4✔
167	leaseBeat = null;	4✔
168	return p;	4✔
169	}
170	}
171	})
172	.then(() => {
173	if (!leaseBeat) {	16✔
174	// Create a new for lease
175	leaseBeat = this.client.lease(this.opts.heartbeatTimeout);	8✔
176
177	//Handle lease-lost event. Release lease when lost. Next heartbeat will request a new lease
178	leaseBeat.on("lost", err => {	8✔
UNCOV 179	this.logger.warn(	×
180	"Lost heartbeat lease. Dropping lease and retrying heartbeat. Error:",
181	err.message
182	);
183	leaseBeat.release();	×
UNCOV 184	this.leaseBeat = null;	×
185	// if broker is connected, send heartbeat immediately. Otherwise it is sent on reconnect.
186	if (this.broker.transit.connected) {	×
UNCOV 187	this.sendHeartbeat();	×
188	}
189	});
190
191	return leaseBeat	8✔
192	.grant() // Waiting for the lease creation on the server
193	.then(() => (this.leaseBeat = leaseBeat));	8✔
194	}
195	})
196	.then(() =>
197	this.leaseBeat.put(key).value(this.serializer.serialize(data, P.PACKET_HEARTBEAT))	16✔
198	)
199	.then(() => (this.lastBeatSeq = seq))	16✔
200	.then(() => this.collectOnlineNodes())	16✔
201	.catch(err => {
202	this.logger.error("Error occurred while collect etcd keys.", err);	4✔
203
204	this.broker.broadcastLocal("$discoverer.error", {	4✔
205	error: err,
206	module: "discoverer",
207	type: C.FAILED_COLLECT_KEYS
208	});
209	})
210	.then(() => {
211	timeEnd();	16✔
212	this.broker.metrics.increment(METRIC.MOLECULER_DISCOVERER_ETCD_COLLECT_TOTAL);	16✔
213	});
214	}
215
216	/**
217	* Collect online nodes from etcd server.
218	*/
219	collectOnlineNodes() {
220	// Get the current node list so that we can check the disconnected nodes.
221	const prevNodes = this.registry.nodes	16✔
222	.list({ onlyAvailable: true, withServices: false })
223	.map(node => node.id)	44✔
224	.filter(nodeID => nodeID !== this.broker.nodeID);	44✔
225
226	// Collect the online node keys.
227	return this.Promise.resolve()	16✔
228	.then(() => {
229	if (this.opts.fullCheck && ++this.idx % this.opts.fullCheck === 0) {	16✔
230	// Full check
231	//this.logger.debug("Full check", this.idx);
232	this.idx = 0;	4✔
233
234	return this.client	4✔
235	.getAll()
236	.prefix(`${this.PREFIX}/beats/`)
237	.buffers()
238	.then(result =>
239	Object.values(result).map(raw => {	4✔
240	try {	12✔
241	return this.serializer.deserialize(raw, P.PACKET_INFO);	12✔
242	} catch (err) {
UNCOV 243	this.logger.warn("Unable to parse HEARTBEAT packet", err, raw);	×
244	}
245	})
246	);
247	} else {
248	//this.logger.debug("Lazy check", this.idx);
249	// Lazy check
250	return this.client	12✔
251	.getAll()
252	.prefix(`${this.PREFIX}/beats/`)
253	.keys()
254	.then(keys =>
255	keys.map(key => {	12✔
256	const p = key.substring(`${this.PREFIX}/beats/`.length).split("/");	12✔
257	return {	12✔
258	key,
259	sender: p[0],
260	instanceID: p[1],
261	seq: Number(p[2])
262	};
263	})
264	);
265	}
266	})
267
268	.then(packets => {
269	_.compact(packets).map(packet => {	16✔
270	if (packet.sender == this.broker.nodeID) return;	24✔
271
272	removeFromArray(prevNodes, packet.sender);	16✔
273	this.heartbeatReceived(packet.sender, packet);	16✔
274	});
275	})
276
277	.then(() => {
278	if (prevNodes.length > 0) {	16✔
279	// Disconnected nodes
280	prevNodes.map(nodeID => {	12✔
281	this.logger.info(	16✔
282	`The node '${nodeID}' is not available. Removing from registry...`
283	);
284	this.remoteNodeDisconnected(nodeID, true);	16✔
285	});
286	}
287	});
288	}
289
290	/**
291	* Discover a new or old node.
292	*
293	* @param {String} nodeID
294	* @returns {Promise<Node \| void>}
295	*/
296	discoverNode(nodeID) {
297	return this.client	12✔
298	.get(`${this.PREFIX}/info/${nodeID}`)
299	.buffer()
300	.then(res => {
301	if (!res) {	12✔
302	this.logger.warn(`No INFO for '${nodeID}' node in registry.`);	4✔
303	return;	4✔
304	}
305	try {	8✔
306	const info = this.serializer.deserialize(res, P.PACKET_INFO);	8✔
307	return this.processRemoteNodeInfo(nodeID, info);	4✔
308	} catch (err) {
309	this.logger.warn("Unable to parse INFO packet", err, res);	4✔
310	}
311	});
312	}
313
314	/**
315	* Discover all nodes (after connected)
316	* @returns {Promise<Node[] \| void>}
317	*/
318	discoverAllNodes() {
319	return this.collectOnlineNodes();	4✔
320	}
321
322	/**
323	* Local service registry has been changed. We should notify remote nodes.
324	* @param {String=} nodeID
325	* @returns {Promise<void>}
326	*/
327	sendLocalNodeInfo(nodeID) {
328	const info = this.broker.getLocalNodeInfo();	24✔
329
330	const payload = Object.assign(	24✔
331	{
332	ver: this.broker.PROTOCOL_VERSION,
333	sender: this.broker.nodeID
334	},
335	info
336	);
337
338	const key = this.INFO_KEY;	24✔
339	const seq = this.localNode.seq;	24✔
340	let leaseInfo = this.leaseInfo;	24✔
341
342	const p =
343	!nodeID && this.broker.options.disableBalancer	24✔
344	? this.transit.tx.makeBalancedSubscriptions()
345	: this.Promise.resolve();
346	return p	24✔
347	.then(() => {
348	if (leaseInfo) {	24✔
349	if (seq != this.lastInfoSeq) {	20✔
350	const p = leaseInfo.revoke();	8✔
351	leaseInfo = null;	8✔
352	return p;	8✔
353	}
354	}
355	})
356	.then(() => {
357	if (!leaseInfo) {	24✔
358	leaseInfo = this.client.lease(60);	12✔
359
360	//Handle lease-lost event. Release lease when lost. Next heartbeat will request a new lease
361	leaseInfo.on("lost", err => {	12✔
362	this.logger.warn(	×
363	"Lost info lease. Dropping lease and retrying info-send. Error:",
364	err.message
365	);
366	leaseInfo.release();	×
UNCOV 367	this.leaseInfo = null;	×
368	// if broker is connected, send local node info immediately. Otherwise it is sent on reconnect.
UNCOV 369	if (this.broker.transit.connected) {	×
UNCOV 370	this.sendLocalNodeInfo(nodeID);	×
371	}
372	});
373
374	return leaseInfo	12✔
375	.grant() // Waiting for the lease creation on the server
376	.then(() => (this.leaseInfo = leaseInfo));	12✔
377	}
378	})
379	.then(() => leaseInfo.put(key).value(this.serializer.serialize(payload, P.PACKET_INFO)))	24✔
380	.then(() => {
381	this.lastInfoSeq = seq;	16✔
382
383	// Sending a new heartbeat because it contains the `seq`
384	if (!nodeID) return this.beat();	16✔
385	})
386	.catch(err => {
387	this.logger.error("Unable to send INFO to etcd server", err);	8✔
388
389	this.broker.broadcastLocal("$discoverer.error", {	8✔
390	error: err,
391	module: "discoverer",
392	type: C.FAILED_SEND_INFO
393	});
394	});
395	}
396
397	/**
398	* Unregister local node after disconnecting.
399	*/
400	localNodeDisconnected() {
401	return this.Promise.resolve()	4✔
402	.then(() => super.localNodeDisconnected())	4✔
403	.then(() => this.logger.debug("Remove local node from registry..."))	4✔
404	.then(() => this.client.delete().key(this.INFO_KEY))	4✔
405	.then(() => this.client.delete().key(this.BEAT_KEY))	4✔
406	.then(() => {
407	if (this.leaseBeat) return this.leaseBeat.revoke();	4!
408	})
409	.then(() => {
410	if (this.leaseInfo) return this.leaseInfo.revoke();	4!
411	});
412	}
413	}
414
415	module.exports = Etcd3Discoverer;	404✔

moleculerjs / moleculer / 7157014227

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