123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211 |
- 'use strict';
- const coreClass = require("../../core");
- const redis = require('redis');
- const config = require('config');
- const mongoose = require('mongoose');
- // Lightweight / convenience wrapper around redis module for our needs
- const pubs = {}, subs = {};
- module.exports = class extends coreClass {
- initialize() {
- return new Promise((resolve, reject) => {
- this.setStage(1);
- this.schemas = {
- session: require('./schemas/session'),
- station: require('./schemas/station'),
- playlist: require('./schemas/playlist'),
- officialPlaylist: require('./schemas/officialPlaylist'),
- song: require('./schemas/song'),
- punishment: require('./schemas/punishment')
- }
- this.url = config.get("redis").url;
- this.password = config.get("redis").password;
- this.logger.info("REDIS", "Connecting...");
- this.client = redis.createClient({
- url: this.url,
- password: this.password,
- retry_strategy: (options) => {
- if (this.state === "LOCKDOWN") return;
- if (this.state !== "RECONNECTING") this.setState("RECONNECTING");
- this.logger.info("CACHE_MODULE", `Attempting to reconnect.`);
- if (options.attempt >= 10) {
- this.logger.error("CACHE_MODULE", `Stopped trying to reconnect.`);
- this.failed = true;
- this._lockdown();
- return undefined;
- }
- return 3000;
- }
- });
- this.client.on('error', err => {
- if (this.state === "INITIALIZING") reject(err);
- if(this.state === "LOCKDOWN") return;
- this.logger.error("CACHE_MODULE", `Error ${err.message}.`);
- });
- this.client.on("connect", () => {
- this.logger.info("CACHE_MODULE", "Connected succesfully.");
- if (this.state === "INITIALIZING") resolve();
- else if (this.state === "LOCKDOWN" || this.state === "RECONNECTING") this.setState("INITIALIZED");
- });
- });
- }
- /**
- * Gracefully closes all the Redis client connections
- */
- async quit() {
- try { await this._validateHook(); } catch { return; }
- if (this.client.connected) {
- this.client.quit();
- Object.keys(pubs).forEach((channel) => pubs[channel].quit());
- Object.keys(subs).forEach((channel) => subs[channel].client.quit());
- }
- }
- /**
- * Sets a single value in a table
- *
- * @param {String} table - name of the table we want to set a key of (table === redis hash)
- * @param {String} key - name of the key to set
- * @param {*} value - the value we want to set
- * @param {Function} cb - gets called when the value has been set in Redis
- * @param {Boolean} [stringifyJson=true] - stringify 'value' if it's an Object or Array
- */
- async hset(table, key, value, cb, stringifyJson = true) {
- try { await this._validateHook(); } catch { return; }
- if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
- // automatically stringify objects and arrays into JSON
- if (stringifyJson && ['object', 'array'].includes(typeof value)) value = JSON.stringify(value);
- this.client.hset(table, key, value, err => {
- if (cb !== undefined) {
- if (err) return cb(err);
- cb(null, JSON.parse(value));
- }
- });
- }
- /**
- * Gets a single value from a table
- *
- * @param {String} table - name of the table to get the value from (table === redis hash)
- * @param {String} key - name of the key to fetch
- * @param {Function} cb - gets called when the value is returned from Redis
- * @param {Boolean} [parseJson=true] - attempt to parse returned data as JSON
- */
- async hget(table, key, cb, parseJson = true) {
- try { await this._validateHook(); } catch { return; }
- if (!key || !table) return typeof cb === 'function' ? cb(null, null) : null;
- if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
- this.client.hget(table, key, (err, value) => {
- if (err) return typeof cb === 'function' ? cb(err) : null;
- if (parseJson) try {
- value = JSON.parse(value);
- } catch (e) {
- }
- if (typeof cb === 'function') cb(null, value);
- });
- }
- /**
- * Deletes a single value from a table
- *
- * @param {String} table - name of the table to delete the value from (table === redis hash)
- * @param {String} key - name of the key to delete
- * @param {Function} cb - gets called when the value has been deleted from Redis or when it returned an error
- */
- async hdel(table, key, cb) {
- try { await this._validateHook(); } catch { return; }
- if (!key || !table || typeof key !== "string") return cb(null, null);
- if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
- this.client.hdel(table, key, (err) => {
- if (err) return cb(err);
- else return cb(null);
- });
- }
- /**
- * Returns all the keys for a table
- *
- * @param {String} table - name of the table to get the values from (table === redis hash)
- * @param {Function} cb - gets called when the values are returned from Redis
- * @param {Boolean} [parseJson=true] - attempts to parse all values as JSON by default
- */
- async hgetall(table, cb, parseJson = true) {
- try { await this._validateHook(); } catch { return; }
- if (!table) return cb(null, null);
- this.client.hgetall(table, (err, obj) => {
- if (err) return typeof cb === 'function' ? cb(err) : null;
- if (parseJson && obj) Object.keys(obj).forEach((key) => { try { obj[key] = JSON.parse(obj[key]); } catch (e) {} });
- if (parseJson && !obj) obj = [];
- cb(null, obj);
- });
- }
- /**
- * Publish a message to a channel, caches the redis client connection
- *
- * @param {String} channel - the name of the channel we want to publish a message to
- * @param {*} value - the value we want to send
- * @param {Boolean} [stringifyJson=true] - stringify 'value' if it's an Object or Array
- */
- async pub(channel, value, stringifyJson = true) {
- try { await this._validateHook(); } catch { return; }
- /*if (pubs[channel] === undefined) {
- pubs[channel] = redis.createClient({ url: this.url });
- pubs[channel].on('error', (err) => console.error);
- }*/
- if (stringifyJson && ['object', 'array'].includes(typeof value)) value = JSON.stringify(value);
- //pubs[channel].publish(channel, value);
- this.client.publish(channel, value);
- }
- /**
- * Subscribe to a channel, caches the redis client connection
- *
- * @param {String} channel - name of the channel to subscribe to
- * @param {Function} cb - gets called when a message is received
- * @param {Boolean} [parseJson=true] - parse the message as JSON
- */
- async sub(channel, cb, parseJson = true) {
- try { await this._validateHook(); } catch { return; }
- if (subs[channel] === undefined) {
- subs[channel] = { client: redis.createClient({ url: this.url, password: this.password }), cbs: [] };
- subs[channel].client.on('message', (channel, message) => {
- if (parseJson) try { message = JSON.parse(message); } catch (e) {}
- subs[channel].cbs.forEach((cb) => cb(message));
- });
- subs[channel].client.subscribe(channel);
- }
- subs[channel].cbs.push(cb);
- }
- }
|