asksuite-core/util/redis.js

const redis = require('redis');
const { promisifyAll } = require('bluebird');
const _ = require('lodash');

promisifyAll(redis.RedisClient.prototype);
promisifyAll(redis.Multi.prototype);

module.exports = configurationParam => {
  const CHANNEL_KEY_UPDATE = '__keyevent@0__:set';

  const configuration = generateConfiguration(configurationParam);
  const client = redis.createClient(configuration);
  const listener = redis.createClient(configuration);
  const databaseListeners = [];
  const channelListeners = [];
  const keyListeners = [];
  const subscribedChannels = [];
  const subscribedKeys = [];

  const clientOnReady = client => {
    const name = configurationParam.name || process.env.PLATFORM;

    if (name) {
      try {
        client.internal_send_command({ command: 'CLIENT', args: ['SETNAME', name] });
      } catch (error) {
        console.error('redis client set name error');
      }
    }

    setInterval(() => client.ping(), 60000);
  };

  const clientOnError = (client, error) => {
    console.error('redis error', error.message);
  };

  client.on('ready', () => clientOnReady(client));
  client.on('error', error => clientOnError(client, error));

  listener.on('ready', () => clientOnReady(listener));
  listener.on('error', error => clientOnError(listener, error));

  function toJSON(value) {
    let cache = [];
    const string = JSON.stringify(value, function(key, value) {
      if (typeof value === 'object' && value !== null) {
        if (cache.indexOf(value) !== -1) {
          // Duplicate reference found, discard key
          return;
        }
        // Store value in our collection
        cache.push(value);
      }
      return value;
    });
    cache = null; // Enable garbage collection

    return string;
  }

  function generateConfiguration(conf) {
    return _.defaults(conf, {
      retry_strategy: () => 1000,
    });
  }

  const delay = ms => new Promise(res => setTimeout(res, ms));

  const isConnected = async () => {
    if (client.connected) {
      return true;
    }

    await delay(1000); // wait for redis connect
    return client.connected;
  };

  const setValue = async (key, value, json = true) => {
    if (!(await isConnected())) {
      return;
    }

    client.setAsync(key, json ? toJSON(value) : value);
  };

  const setValueWithExpirationTime = async (key, value, json = true, expirationTime) => {
    if (!(await isConnected())) {
      return;
    }
    client.setAsync(key, json ? toJSON(value) : value, 'EX', expirationTime);
  };

  const getValue = async (key, json = true) => {
    try {
      // if (!await isConnected()) {
      //     return null;
      // }

      const value = await client.getAsync(key);
      return json ? JSON.parse(value) : value;
    } catch (error) {
      return null;
    }
  };
  const mergeValue = async (key, value, json = true) => {
    const oldValue = await getValue(key);
    const mergedValue = Object.assign({}, oldValue, value);
    return setValue(key, mergedValue, json);
  };

  const expire = (key, time) => {
    client.expire(key, time);
  };

  const remove = key => {
    expire(key, 1);
  };

  const publish = (channel, message) => client.publishAsync(channel, message);
  const alreadySubscribedChannel = channel =>
    subscribedChannels.find(_channel => _channel === channel);
  const alreadySubscribedKey = key => subscribedKeys.find(_key => _key === key);

  const registerDatabaseListener = processMessage => {
    listener.subscribe(CHANNEL_KEY_UPDATE);
    databaseListeners.push({ channel: CHANNEL_KEY_UPDATE, processMessage });
  };

  const registerChannelListener = (channel, processMessage) => {
    if (!alreadySubscribedChannel(channel)) {
      subscribedChannels.push(channel);
      listener.subscribe(channel);
    }
    channelListeners.push({ channel, processMessage });
  };

  const registerKeyListener = (key, processMessage) => {
    if (!alreadySubscribedKey(key)) {
      subscribedKeys.push(key);
      listener.subscribe(CHANNEL_KEY_UPDATE, key);
    }
    keyListeners.push({ channel: CHANNEL_KEY_UPDATE, key, processMessage });
  };

  listener.on('message', async (channel, message) => {
    let registeredListeners = [];
    const messageKey = message;

    if (channel === CHANNEL_KEY_UPDATE) {
      registeredListeners = keyListeners.filter(keyListener => keyListener.key === messageKey);
      registeredListeners = registeredListeners.concat(databaseListeners);

      message = await getValue(messageKey);
    } else {
      registeredListeners = channelListeners.filter(listener => listener.channel === channel);
    }

    registeredListeners.forEach(({ channel, key, processMessage }) =>
      processMessage(message, key || messageKey, channel),
    );
  });

  return {
    setValue,
    setValueWithExpirationTime,
    getValue,
    remove,
    expire,
    mergeValue,
    publish,
    registerChannelListener,
    registerKeyListener,
    registerDatabaseListener,
  };
};