notifications.js 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106
  1. 'use strict';
  2. const crypto = require('crypto');
  3. const redis = require('redis');
  4. const logger = require('./logger');
  5. const subscriptions = [];
  6. let initialized = false;
  7. let lockdown = false;
  8. let errorCb;
  9. const lib = {
  10. pub: null,
  11. sub: null,
  12. errorCb: null,
  13. /**
  14. * Initializes the notifications module
  15. *
  16. * @param {String} url - the url of the redis server
  17. * @param {String} password - the password of the redis server
  18. * @param {Function} cb - gets called once we're done initializing
  19. */
  20. init: (url, password, errorCb, cb) => {
  21. lib.errorCb = errorCb;
  22. lib.pub = redis.createClient({ url, password });
  23. lib.sub = redis.createClient({ url, password });
  24. lib.sub.on('error', (err) => {
  25. errorCb('Cache connection error.', err, 'Notifications');
  26. });
  27. lib.sub.on('pmessage', (pattern, channel, expiredKey) => {
  28. logger.stationIssue(`PMESSAGE - Pattern: ${pattern}; Channel: ${channel}; ExpiredKey: ${expiredKey}`);
  29. subscriptions.forEach((sub) => {
  30. if (sub.name !== expiredKey) return;
  31. sub.cb();
  32. });
  33. });
  34. lib.sub.psubscribe('__keyevent@0__:expired');
  35. initialized = true;
  36. if (lockdown) return this._lockdown();
  37. cb();
  38. },
  39. /**
  40. * Schedules a notification to be dispatched in a specific amount of milliseconds,
  41. * notifications are unique by name, and the first one is always kept, as in
  42. * attempting to schedule a notification that already exists won't do anything
  43. *
  44. * @param {String} name - the name of the notification we want to schedule
  45. * @param {Integer} time - how long in milliseconds until the notification should be fired
  46. * @param {Function} cb - gets called when the notification has been scheduled
  47. */
  48. schedule: (name, time, cb, station) => {
  49. if (lockdown) return;
  50. if (!cb) cb = ()=>{};
  51. time = Math.round(time);
  52. logger.stationIssue(`SCHEDULE - Time: ${time}; Name: ${name}; Key: ${crypto.createHash('md5').update(`_notification:${name}_`).digest('hex')}; StationId: ${station._id}; StationName: ${station.name}`);
  53. lib.pub.set(crypto.createHash('md5').update(`_notification:${name}_`).digest('hex'), '', 'PX', time, 'NX', cb);
  54. },
  55. /**
  56. * Subscribes a callback function to be called when a notification gets called
  57. *
  58. * @param {String} name - the name of the notification we want to subscribe to
  59. * @param {Function} cb - gets called when the subscribed notification gets called
  60. * @param {Boolean} unique - only subscribe if another subscription with the same name doesn't already exist
  61. * @return {Object} - the subscription object
  62. */
  63. subscribe: (name, cb, unique = false, station) => {
  64. if (lockdown) return;
  65. logger.stationIssue(`SUBSCRIBE - Name: ${name}; Key: ${crypto.createHash('md5').update(`_notification:${name}_`).digest('hex')}, StationId: ${station._id}; StationName: ${station.name}; Unique: ${unique}; SubscriptionExists: ${!!subscriptions.find((subscription) => subscription.originalName == name)};`);
  66. if (unique && !!subscriptions.find((subscription) => subscription.originalName == name)) return;
  67. let subscription = { originalName: name, name: crypto.createHash('md5').update(`_notification:${name}_`).digest('hex'), cb };
  68. subscriptions.push(subscription);
  69. return subscription;
  70. },
  71. /**
  72. * Remove a notification subscription
  73. *
  74. * @param {Object} subscription - the subscription object returned by {@link subscribe}
  75. */
  76. remove: (subscription) => {
  77. if (lockdown) return;
  78. let index = subscriptions.indexOf(subscription);
  79. if (index) subscriptions.splice(index, 1);
  80. },
  81. unschedule: (name) => {
  82. if (lockdown) return;
  83. logger.stationIssue(`UNSCHEDULE - Name: ${name}; Key: ${crypto.createHash('md5').update(`_notification:${name}_`).digest('hex')}`);
  84. lib.pub.del(crypto.createHash('md5').update(`_notification:${name}_`).digest('hex'));
  85. },
  86. _lockdown: () => {
  87. lib.pub.quit();
  88. lib.sub.quit();
  89. lockdown = true;
  90. }
  91. };
  92. module.exports = lib;