#1892

Build # #1892

Build Type

push

Committed by

Commit Message

Merge pull request #1192 from shawnmcknight/redlock-opt-out

Allow opt out of redlock in redis cacher

Run Details

3965 of 4484 branches covered (88.43%)

Branch coverage included in aggregate %.

16 of 16 new or added lines in 1 file covered. (100.0%)

7337 of 7557 relevant lines covered (97.09%)

248.27 hits per line

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

88.48

/src/cachers/redis.js

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

"use strict";

let Redis;
const BaseCacher = require("./base");
const _ = require("lodash");
const { METRIC } = require("../metrics");
const { BrokerOptionsError } = require("../errors");
const Serializers = require("../serializers");
const C = require("../constants");

/**
 * Cacher factory for Redis
 *
 * @class RedisCacher
 */
class RedisCacher extends BaseCacher {
        /**
         * Creates an instance of RedisCacher.
         *
         * @param {object} opts
         *
         * @memberof RedisCacher
         */
        constructor(opts) {
                if (typeof opts === "string") opts = { redis: opts };

                super(opts);

                this.opts = _.defaultsDeep(this.opts, {
                        prefix: null,
                        pingInterval: null
                });

                this.pingIntervalHandle = null;

                /**
                 * Redlock client instance
                 * @memberof RedisCacher
                 */
                this.redlock = null;
                this.redlockNonBlocking = null;
        }

        /**
         * Initialize cacher. Connect to Redis server
         *
         * @param {any} broker
         *
         * @memberof RedisCacher
         */
        init(broker) {
                super.init(broker);
                try {
                        Redis = require("ioredis");
                } catch (err) {
                        /* istanbul ignore next */
                        this.broker.fatal(
                                "The 'ioredis' package is missing. Please install it with 'npm install ioredis --save' command.",
                                err,
                                true
                        );
                }
                /**
                 * ioredis client instance
                 * @memberof RedisCacher
                 */
                if (this.opts.cluster) {
                        if (!this.opts.cluster.nodes || this.opts.cluster.nodes.length === 0) {
                                throw new BrokerOptionsError("No nodes defined for cluster");
                        }

                        this.client = new Redis.Cluster(this.opts.cluster.nodes, this.opts.cluster.options);
                } else {
                        this.client = new Redis(this.opts.redis);
                }

                this.connected = false;

                this.client.on("connect", () => {
                        this.connected = true;

                        /* istanbul ignore next */
                        this.logger.info("Redis cacher connected.");
                });

                this.client.on("error", err => {
                        this.connected = false;

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

                        /* istanbul ignore next */
                        this.logger.error(err);
                });

                // check for !== false for backwards compatibility purposes; should be changed to opt-in in a breaking change release
                if (this.opts.redlock !== false) {
                        let Redlock;
                        try {
                                Redlock = require("redlock");
                        } catch (err) {
                                Redlock = null;
                        }
                        if (Redlock != null) {
                                let redlockClients = (this.opts.redlock ? this.opts.redlock.clients : null) || [
                                        this.client
                                ];

                                this.redlock = new Redlock(redlockClients, _.omit(this.opts.redlock, ["clients"]));
                                // Non-blocking redlock client, used for tryLock()
                                this.redlockNonBlocking = new Redlock(redlockClients, {
                                        retryCount: 0
                                });
                        }
                }
                if (this.opts.monitor) {
                        /* istanbul ignore next */
                        this.client.monitor((err, monitor) => {
                                this.logger.debug("Redis cacher entering monitoring mode...");
                                monitor.on("monitor", (time, args /*, source, database*/) => {
                                        this.logger.debug(args);
                                });
                        });
                }

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

                // add interval for ping if set
                if (this.opts.pingInterval > 0) {
                        this.pingIntervalHandle = setInterval(() => {
                                this.client
                                        .ping()
                                        .then(() => {
                                                this.connected = true;

                                                this.logger.debug("Sent PING to Redis Server");
                                        })
                                        .catch(err => {
                                                this.connected = false;

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

                                                this.logger.error("Failed to send PING to Redis Server", err);
                                        });
                        }, Number(this.opts.pingInterval));
                }

                this.logger.debug("Redis Cacher created. Prefix: " + this.prefix);
        }

        /**
         * Close Redis client connection
         *
         * @memberof RedisCacher
         */
        close() {
                if (this.pingIntervalHandle != null) {
                        clearInterval(this.pingIntervalHandle);
                        this.pingIntervalHandle = null;
                }
                return this.client != null ? this.client.quit() : Promise.resolve();
        }

        /**
         * Get data from cache by key
         *
         * @param {any} key
         * @returns {Promise}
         *
         * @memberof Cacher
         */
        get(key) {
                this.logger.debug(`GET ${key}`);
                this.metrics.increment(METRIC.MOLECULER_CACHER_GET_TOTAL);
                const timeEnd = this.metrics.timer(METRIC.MOLECULER_CACHER_GET_TIME);

                return this.client.getBuffer(this.prefix + key).then(data => {
                        if (data) {
                                this.logger.debug(`FOUND ${key}`);
                                this.metrics.increment(METRIC.MOLECULER_CACHER_FOUND_TOTAL);

                                try {
                                        const res = this.serializer.deserialize(data);
                                        timeEnd();

                                        return res;
                                } catch (err) {
                                        this.logger.error("Redis result parse error.", err, data);
                                }
                        }
                        timeEnd();
                        return null;
                });
        }

        /**
         * Save data to cache by key
         *
         * @param {String} key
         * @param {any} data JSON object
         * @param {Number} ttl Optional Time-to-Live
         * @returns {Promise}
         *
         * @memberof Cacher
         */
        set(key, data, ttl) {
                this.metrics.increment(METRIC.MOLECULER_CACHER_SET_TOTAL);
                const timeEnd = this.metrics.timer(METRIC.MOLECULER_CACHER_SET_TIME);

                data = this.serializer.serialize(data);
                this.logger.debug(`SET ${key}`);

                if (ttl == null) ttl = this.opts.ttl;

                let p;
                if (ttl) {
                        p = this.client.setex(this.prefix + key, ttl, data);
                } else {
                        p = this.client.set(this.prefix + key, data);
                }

                return p
                        .then(res => {
                                timeEnd();
                                return res;
                        })
                        .catch(err => {
                                timeEnd();
                                throw err;
                        });
        }

        /**
         * Delete a key from cache
         *
         * @param {string|Array<string>} deleteTargets
         * @returns {Promise}
         *
         * @memberof Cacher
         */
        del(deleteTargets) {
                this.metrics.increment(METRIC.MOLECULER_CACHER_DEL_TOTAL);
                const timeEnd = this.metrics.timer(METRIC.MOLECULER_CACHER_DEL_TIME);

                deleteTargets = Array.isArray(deleteTargets) ? deleteTargets : [deleteTargets];
                const keysToDelete = deleteTargets.map(key => this.prefix + key);
                this.logger.debug(`DELETE ${keysToDelete}`);
                return this.client
                        .del(keysToDelete)
                        .then(res => {
                                timeEnd();
                                return res;
                        })
                        .catch(err => {
                                timeEnd();
                                this.logger.error(`Redis 'del' error. Key: ${keysToDelete}`, err);
                                throw err;
                        });
        }

        /**
         * Clean cache. Remove every key by prefix
         *        http://stackoverflow.com/questions/4006324/how-to-atomically-delete-keys-matching-a-pattern-using-redis
         * alternative solution:
         *        https://github.com/cayasso/cacheman-redis/blob/master/lib/index.js#L125
         * @param {String|Array<String>} match Match string for SCAN. Default is "*"
         * @returns {Promise}
         *
         * @memberof Cacher
         */
        clean(match = "*") {
                this.metrics.increment(METRIC.MOLECULER_CACHER_CLEAN_TOTAL);
                const timeEnd = this.metrics.timer(METRIC.MOLECULER_CACHER_CLEAN_TIME);

                const cleaningPatterns = Array.isArray(match) ? match : [match];
                const normalizedPatterns = cleaningPatterns.map(
                        match => this.prefix + match.replace(/\*\*/g, "*")
                );
                this.logger.debug(`CLEAN ${match}`);
                return this._sequentialPromises(normalizedPatterns)
                        .then(res => {
                                timeEnd();
                                return res;
                        })
                        .catch(err => {
                                timeEnd();
                                this.logger.error(`Redis 'scanDel' error. Pattern: ${err.pattern}`, err);
                                throw err;
                        });
        }

        /**
         * Get data and ttl from cache by key.
         *
         * @param {string|Array<string>} key
         * @returns {Promise}
         *
         * @memberof RedisCacher
         */
        getWithTTL(key) {
                return this.client
                        .pipeline()
                        .getBuffer(this.prefix + key)
                        .ttl(this.prefix + key)
                        .exec()
                        .then(res => {
                                let [err0, data] = res[0];
                                let [err1, ttl] = res[1];
                                if (err0) {
                                        return this.broker.Promise.reject(err0);
                                }
                                if (err1) {
                                        return this.broker.Promise.reject(err1);
                                }
                                if (data) {
                                        this.logger.debug(`FOUND ${key}`);
                                        try {
                                                data = this.serializer.deserialize(data);
                                        } catch (err) {
                                                this.logger.error("Redis result parse error.", err, data);
                                                data = null;
                                        }
                                }
                                return { data, ttl };
                        });
        }

        /**
         * Acquire a lock
         *
         * @param {string|Array<string>} key
         * @param {Number} ttl Optional Time-to-Live
         * @returns {Promise}
         *
         * @memberof RedisCacher
         */
        lock(key, ttl = 15000) {
                if (this.redlock == null) {
                        return this._handleMissingRedlock();
                }

                key = this.prefix + key + "-lock";
                return this.redlock.lock(key, ttl).then(lock => {
                        return () => lock.unlock();
                });
        }

        /**
         * Try to acquire a lock
         *
         * @param {string|Array<string>} key
         * @param {Number} ttl Optional Time-to-Live
         * @returns {Promise}
         *
         * @memberof RedisCacher
         */
        tryLock(key, ttl = 15000) {
                if (this.redlockNonBlocking == null) {
                        return this._handleMissingRedlock();
                }

                key = this.prefix + key + "-lock";
                return this.redlockNonBlocking.lock(key, ttl).then(lock => {
                        return () => lock.unlock();
                });
        }

        /**
         * Common code for handling unavailable Redlock
         * @returns {Promise}
         */
        _handleMissingRedlock() {
                this.logger.error(
                        "The 'redlock' package is missing or redlock is disabled. If you want to enable cache lock, please install it with 'npm install redlock --save' command."
                );
                return Promise.resolve();
        }

        _sequentialPromises(elements) {
                return elements.reduce((chain, element) => {
                        return chain.then(() => this._scanDel(element));
                }, this.broker.Promise.resolve());
        }

        _clusterScanDel(pattern) {
                const scanDelPromises = [];
                // get only master nodes to scan for deletion,
                // if we get slave nodes, it would be failed for deletion.
                const nodes = this.client.nodes("master");

                nodes.forEach(node => {
                        scanDelPromises.push(this._nodeScanDel(node, pattern));
                });

                return this.broker.Promise.all(scanDelPromises);
        }

        _nodeScanDel(node, pattern) {
                return new Promise((resolve, reject) => {
                        const stream = node.scanStream({
                                match: pattern,
                                count: 100
                        });

                        stream.on("data", (keys = []) => {
                                if (!keys.length) {
                                        return;
                                }

                                stream.pause();
                                node.del(keys)
                                        .then(() => {
                                                stream.resume();
                                        })
                                        .catch(err => {
                                                err.pattern = pattern;
                                                return reject(err);
                                        });
                        });

                        stream.on("error", err => {
                                this.logger.error(`Error occured while deleting keys '${pattern}' from node.`, err);
                                reject(err);
                        });

                        stream.on("end", () => {
                                // End deleting keys from node
                                resolve();
                        });
                });
        }

        _scanDel(pattern) {
                if (this.client instanceof Redis.Cluster) {
                        return this._clusterScanDel(pattern);
                } else {
                        return this._nodeScanDel(this.client, pattern);
                }
        }

        /**
         * Return all cache keys with available properties (ttl, lastUsed, ...etc).
         *
         * @returns Promise<Array<Object>>
         */
        getCacheKeys() {
                return new Promise((resolve, reject) => {
                        const res = [];

                        const stream = this.client.scanStream({
                                match: this.prefix + "*",
                                count: 100
                        });

                        stream.on("data", (keys = []) => res.push(...keys));

                        stream.on("error", err => {
                                this.logger.error("Error occured while listing keys from node.", err);
                                reject(err);
                        });

                        stream.on("end", () => {
                                // End deleting keys from node
                                resolve(
                                        res.map(key => ({
                                                key: key.startsWith(this.prefix) ? key.slice(this.prefix.length) : key
                                        }))
                                );
                        });
                });
        }
}

module.exports = RedisCacher;

1	/*
2	* moleculer
3	* Copyright (c) 2020 MoleculerJS (https://github.com/moleculerjs/moleculer)
4	* MIT Licensed
5	*/
6
7	"use strict";
8
9	let Redis;
10	const BaseCacher = require("./base");	103✔
11	const _ = require("lodash");	103✔
12	const { METRIC } = require("../metrics");	103✔
13	const { BrokerOptionsError } = require("../errors");	103✔
14	const Serializers = require("../serializers");	103✔
15	const C = require("../constants");	103✔
16
17	/**
18	* Cacher factory for Redis
19	*
20	* @class RedisCacher
21	*/
22	class RedisCacher extends BaseCacher {
23	/**
24	* Creates an instance of RedisCacher.
25	*
26	* @param {object} opts
27	*
28	* @memberof RedisCacher
29	*/
30	constructor(opts) {
31	if (typeof opts === "string") opts = { redis: opts };	27✔
32
33	super(opts);	27✔
34
35	this.opts = _.defaultsDeep(this.opts, {	27✔
36	prefix: null,
37	pingInterval: null
38	});
39
40	this.pingIntervalHandle = null;	27✔
41
42	/**
43	* Redlock client instance
44	* @memberof RedisCacher
45	*/
46	this.redlock = null;	27✔
47	this.redlockNonBlocking = null;	27✔
48	}
49
50	/**
51	* Initialize cacher. Connect to Redis server
52	*
53	* @param {any} broker
54	*
55	* @memberof RedisCacher
56	*/
57	init(broker) {
58	super.init(broker);	18✔
59	try {	18✔
60	Redis = require("ioredis");	18✔
61	} catch (err) {
62	/* istanbul ignore next */
63	this.broker.fatal(
64	"The 'ioredis' package is missing. Please install it with 'npm install ioredis --save' command.",
65	err,
66	true
67	);
68	}
69	/**
70	* ioredis client instance
71	* @memberof RedisCacher
72	*/
73	if (this.opts.cluster) {	18✔
74	if (!this.opts.cluster.nodes \|\| this.opts.cluster.nodes.length === 0) {	2✔
75	throw new BrokerOptionsError("No nodes defined for cluster");	1✔
76	}
77
78	this.client = new Redis.Cluster(this.opts.cluster.nodes, this.opts.cluster.options);	1✔
79	} else {
80	this.client = new Redis(this.opts.redis);	16✔
81	}
82
83	this.connected = false;	17✔
84
85	this.client.on("connect", () => {	17✔
86	this.connected = true;	×
87
88	/* istanbul ignore next */
89	this.logger.info("Redis cacher connected.");
90	});
91
92	this.client.on("error", err => {	17✔
93	this.connected = false;	1✔
94
95	this.broker.broadcastLocal("$cacher.error", {	1✔
96	error: err,
97	module: "cacher",
98	type: C.CLIENT_ERROR
99	});
100
101	/* istanbul ignore next */
102	this.logger.error(err);
103	});
104
105	// check for !== false for backwards compatibility purposes; should be changed to opt-in in a breaking change release
106	if (this.opts.redlock !== false) {	17✔
107	let Redlock;
108	try {	16✔
109	Redlock = require("redlock");	16✔
110	} catch (err) {
111	Redlock = null;	×
112	}
113	if (Redlock != null) {	16!
114	let redlockClients = (this.opts.redlock ? this.opts.redlock.clients : null) \|\| [	16!
115	this.client
116	];
117
118	this.redlock = new Redlock(redlockClients, _.omit(this.opts.redlock, ["clients"]));	16✔
119	// Non-blocking redlock client, used for tryLock()
120	this.redlockNonBlocking = new Redlock(redlockClients, {	16✔
121	retryCount: 0
122	});
123	}
124	}
125	if (this.opts.monitor) {	17!
126	/* istanbul ignore next */
127	this.client.monitor((err, monitor) => {
128	this.logger.debug("Redis cacher entering monitoring mode...");
129	monitor.on("monitor", (time, args /, source, database/) => {
130	this.logger.debug(args);
131	});
132	});
133	}
134
135	// create an instance of serializer (default to JSON)
136	this.serializer = Serializers.resolve(this.opts.serializer);	17✔
137	this.serializer.init(this.broker);	17✔
138
139	// add interval for ping if set
140	if (this.opts.pingInterval > 0) {	17✔
141	this.pingIntervalHandle = setInterval(() => {	3✔
142	this.client	6✔
143	.ping()
144	.then(() => {
145	this.connected = true;	6✔
146
147	this.logger.debug("Sent PING to Redis Server");	6✔
148	})
149	.catch(err => {
150	this.connected = false;	×
151
152	this.broker.broadcastLocal("$cacher.error", {	×
153	error: err,
154	module: "cacher",
155	type: C.FAILED_SEND_PING
156	});
157
158	this.logger.error("Failed to send PING to Redis Server", err);	×
159	});
160	}, Number(this.opts.pingInterval));
161	}
162
163	this.logger.debug("Redis Cacher created. Prefix: " + this.prefix);	17✔
164	}
165
166	/**
167	* Close Redis client connection
168	*
169	* @memberof RedisCacher
170	*/
171	close() {
172	if (this.pingIntervalHandle != null) {	6✔
173	clearInterval(this.pingIntervalHandle);	1✔
174	this.pingIntervalHandle = null;	1✔
175	}
176	return this.client != null ? this.client.quit() : Promise.resolve();	6✔
177	}
178
179	/**
180	* Get data from cache by key
181	*
182	* @param {any} key
183	* @returns {Promise}
184	*
185	* @memberof Cacher
186	*/
187	get(key) {
188	this.logger.debug(`GET ${key}`);	3✔
189	this.metrics.increment(METRIC.MOLECULER_CACHER_GET_TOTAL);	3✔
190	const timeEnd = this.metrics.timer(METRIC.MOLECULER_CACHER_GET_TIME);	3✔
191
192	return this.client.getBuffer(this.prefix + key).then(data => {	3✔
193	if (data) {	3✔
194	this.logger.debug(`FOUND ${key}`);	2✔
195	this.metrics.increment(METRIC.MOLECULER_CACHER_FOUND_TOTAL);	2✔
196
197	try {	2✔
198	const res = this.serializer.deserialize(data);	2✔
199	timeEnd();	1✔
200
201	return res;	1✔
202	} catch (err) {
203	this.logger.error("Redis result parse error.", err, data);	1✔
204	}
205	}
206	timeEnd();	2✔
207	return null;	2✔
208	});
209	}
210
211	/**
212	* Save data to cache by key
213	*
214	* @param {String} key
215	* @param {any} data JSON object
216	* @param {Number} ttl Optional Time-to-Live
217	* @returns {Promise}
218	*
219	* @memberof Cacher
220	*/
221	set(key, data, ttl) {
222	this.metrics.increment(METRIC.MOLECULER_CACHER_SET_TOTAL);	2✔
223	const timeEnd = this.metrics.timer(METRIC.MOLECULER_CACHER_SET_TIME);	2✔
224
225	data = this.serializer.serialize(data);	2✔
226	this.logger.debug(`SET ${key}`);	2✔
227
228	if (ttl == null) ttl = this.opts.ttl;	2!
229
230	let p;
231	if (ttl) {	2✔
232	p = this.client.setex(this.prefix + key, ttl, data);	1✔
233	} else {
234	p = this.client.set(this.prefix + key, data);	1✔
235	}
236
237	return p	2✔
238	.then(res => {
239	timeEnd();	2✔
240	return res;	2✔
241	})
242	.catch(err => {
243	timeEnd();	×
244	throw err;	×
245	});
246	}
247
248	/**
249	* Delete a key from cache
250	*
251	* @param {string\|Array<string>} deleteTargets
252	* @returns {Promise}
253	*
254	* @memberof Cacher
255	*/
256	del(deleteTargets) {
257	this.metrics.increment(METRIC.MOLECULER_CACHER_DEL_TOTAL);	5✔
258	const timeEnd = this.metrics.timer(METRIC.MOLECULER_CACHER_DEL_TIME);	5✔
259
260	deleteTargets = Array.isArray(deleteTargets) ? deleteTargets : [deleteTargets];	5✔
261	const keysToDelete = deleteTargets.map(key => this.prefix + key);	7✔
262	this.logger.debug(`DELETE ${keysToDelete}`);	5✔
263	return this.client	5✔
264	.del(keysToDelete)
265	.then(res => {
266	timeEnd();	4✔
267	return res;	4✔
268	})
269	.catch(err => {
270	timeEnd();	1✔
271	this.logger.error(`Redis 'del' error. Key: ${keysToDelete}`, err);	1✔
272	throw err;	1✔
273	});
274	}
275
276	/**
277	* Clean cache. Remove every key by prefix
278	* http://stackoverflow.com/questions/4006324/how-to-atomically-delete-keys-matching-a-pattern-using-redis
279	* alternative solution:
280	* https://github.com/cayasso/cacheman-redis/blob/master/lib/index.js#L125
281	* @param {String\|Array<String>} match Match string for SCAN. Default is "*"
282	* @returns {Promise}
283	*
284	* @memberof Cacher
285	*/
286	clean(match = "*") {	3✔
287	this.metrics.increment(METRIC.MOLECULER_CACHER_CLEAN_TOTAL);	7✔
288	const timeEnd = this.metrics.timer(METRIC.MOLECULER_CACHER_CLEAN_TIME);	7✔
289
290	const cleaningPatterns = Array.isArray(match) ? match : [match];	7✔
291	const normalizedPatterns = cleaningPatterns.map(	7✔
292	match => this.prefix + match.replace(/\\/g, "*")	9✔
293	);
294	this.logger.debug(`CLEAN ${match}`);	7✔
295	return this._sequentialPromises(normalizedPatterns)	7✔
296	.then(res => {
297	timeEnd();	6✔
298	return res;	6✔
299	})
300	.catch(err => {
301	timeEnd();	1✔
302	this.logger.error(`Redis 'scanDel' error. Pattern: ${err.pattern}`, err);	1✔
303	throw err;	1✔
304	});
305	}
306
307	/**
308	* Get data and ttl from cache by key.
309	*
310	* @param {string\|Array<string>} key
311	* @returns {Promise}
312	*
313	* @memberof RedisCacher
314	*/
315	getWithTTL(key) {
316	return this.client	4✔
317	.pipeline()
318	.getBuffer(this.prefix + key)
319	.ttl(this.prefix + key)
320	.exec()
321	.then(res => {
322	let [err0, data] = res[0];	4✔
323	let [err1, ttl] = res[1];	4✔
324	if (err0) {	4✔
325	return this.broker.Promise.reject(err0);	1✔
326	}
327	if (err1) {	3✔
328	return this.broker.Promise.reject(err1);	1✔
329	}
330	if (data) {	2!
331	this.logger.debug(`FOUND ${key}`);	2✔
332	try {	2✔
333	data = this.serializer.deserialize(data);	2✔
334	} catch (err) {
335	this.logger.error("Redis result parse error.", err, data);	1✔
336	data = null;	1✔
337	}
338	}
339	return { data, ttl };	2✔
340	});
341	}
342
343	/**
344	* Acquire a lock
345	*
346	* @param {string\|Array<string>} key
347	* @param {Number} ttl Optional Time-to-Live
348	* @returns {Promise}
349	*
350	* @memberof RedisCacher
351	*/
352	lock(key, ttl = 15000) {	1✔
353	if (this.redlock == null) {	3✔
354	return this._handleMissingRedlock();	1✔
355	}
356
357	key = this.prefix + key + "-lock";	2✔
358	return this.redlock.lock(key, ttl).then(lock => {	2✔
359	return () => lock.unlock();	2✔
360	});
361	}
362
363	/**
364	* Try to acquire a lock
365	*
366	* @param {string\|Array<string>} key
367	* @param {Number} ttl Optional Time-to-Live
368	* @returns {Promise}
369	*
370	* @memberof RedisCacher
371	*/
372	tryLock(key, ttl = 15000) {	1✔
373	if (this.redlockNonBlocking == null) {	4✔
374	return this._handleMissingRedlock();	1✔
375	}
376
377	key = this.prefix + key + "-lock";	3✔
378	return this.redlockNonBlocking.lock(key, ttl).then(lock => {	3✔
379	return () => lock.unlock();	2✔
380	});
381	}
382
383	/**
384	* Common code for handling unavailable Redlock
385	* @returns {Promise}
386	*/
387	_handleMissingRedlock() {
388	this.logger.error(	2✔
389	"The 'redlock' package is missing or redlock is disabled. If you want to enable cache lock, please install it with 'npm install redlock --save' command."
390	);
391	return Promise.resolve();	2✔
392	}
393
394	_sequentialPromises(elements) {
395	return elements.reduce((chain, element) => {	7✔
396	return chain.then(() => this._scanDel(element));	9✔
397	}, this.broker.Promise.resolve());
398	}
399
400	_clusterScanDel(pattern) {
401	const scanDelPromises = [];	×
402	// get only master nodes to scan for deletion,
403	// if we get slave nodes, it would be failed for deletion.
404	const nodes = this.client.nodes("master");	×
405
406	nodes.forEach(node => {	×
407	scanDelPromises.push(this._nodeScanDel(node, pattern));	×
408	});
409
410	return this.broker.Promise.all(scanDelPromises);	×
411	}
412
413	_nodeScanDel(node, pattern) {
414	return new Promise((resolve, reject) => {	8✔
415	const stream = node.scanStream({	8✔
416	match: pattern,
417	count: 100
418	});
419
420	stream.on("data", (keys = []) => {	8!
421	if (!keys.length) {	8✔
422	return;	1✔
423	}
424
425	stream.pause();	7✔
426	node.del(keys)	7✔
427	.then(() => {
428	stream.resume();	6✔
429	})
430	.catch(err => {
431	err.pattern = pattern;	1✔
432	return reject(err);	1✔
433	});
434	});
435
436	stream.on("error", err => {	8✔
437	this.logger.error(`Error occured while deleting keys '${pattern}' from node.`, err);	×
438	reject(err);	×
439	});
440
441	stream.on("end", () => {	8✔
442	// End deleting keys from node
443	resolve();	7✔
444	});
445	});
446	}
447
448	_scanDel(pattern) {
449	if (this.client instanceof Redis.Cluster) {	8!
450	return this._clusterScanDel(pattern);	×
451	} else {
452	return this._nodeScanDel(this.client, pattern);	8✔
453	}
454	}
455
456	/**
457	* Return all cache keys with available properties (ttl, lastUsed, ...etc).
458	*
459	* @returns Promise<Array<Object>>
460	*/
461	getCacheKeys() {
462	return new Promise((resolve, reject) => {	1✔
463	const res = [];	1✔
464
465	const stream = this.client.scanStream({	1✔
466	match: this.prefix + "*",
467	count: 100
468	});
469
470	stream.on("data", (keys = []) => res.push(...keys));	1!
471
472	stream.on("error", err => {	1✔
473	this.logger.error("Error occured while listing keys from node.", err);	×
474	reject(err);	×
475	});
476
477	stream.on("end", () => {	1✔
478	// End deleting keys from node
479	resolve(	1✔
480	res.map(key => ({	3✔
481	key: key.startsWith(this.prefix) ? key.slice(this.prefix.length) : key	3✔
482	}))
483	);
484	});
485	});
486	}
487	}
488
489	module.exports = RedisCacher;	103✔

moleculerjs / moleculer / #1892

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