src/pubsub/redis.js

   1 import { EventEmitter } from 'events';
   2 import { v4 as uuidv4 } from 'uuid';
   3
   4 const LOGGER = require('@calzoneman/jsli')('redis-messagebus');
   5
   6 class RedisMessageBus extends EventEmitter {
   7     constructor(pubClient, subClient, channel) {
   8         super();
   9
  10         this.pubClient = pubClient;
  11         this.subClient = subClient;
  12         this.channel = channel;
  13         this.publisherID = uuidv4();
  14
  15         subClient.once('ready', this.subscribe.bind(this));
  16     }
  17
  18     subscribe() {
  19         this.subClient.subscribe(this.channel);
  20         this.subClient.on('message', this.onMessage.bind(this));
  21
  22         LOGGER.info('Subscribed to Redis messages on channel %s', this.channel);
  23     }
  24
  25     onMessage(channel, message) {
  26         if (channel !== this.channel) {
  27             LOGGER.warn('Ignoring message from mismatched channel "%s"', channel);
  28             return;
  29         }
  30
  31         try {
  32             const { event, payload } = JSON.parse(message);
  33
  34             this._emit(event, payload);
  35         } catch (error) {
  36             if (error instanceof SyntaxError) {
  37                 LOGGER.error(
  38                         'Malformed message received: %s (message: "%s")',
  39                         message,
  40                         error
  41                 );
  42             } else {
  43                 LOGGER.error('Unexpected error decoding message: %s', error.stack);
  44             }
  45
  46             return;
  47         }
  48     }
  49
  50     async emit(event, payload) {
  51         try {
  52             const message = JSON.stringify({
  53                 time: new Date(),
  54                 publisher: this.publisherID,
  55                 event,
  56                 payload
  57             });
  58
  59             await this.pubClient.publish(this.channel, message);
  60         } catch (error) {
  61             LOGGER.error('Unable to send event %s: %s', event, error);
  62         }
  63     }
  64 }
  65
  66 Object.assign(RedisMessageBus.prototype, {
  67     _emit: EventEmitter.prototype.emit
  68 });
  69
  70 export { RedisMessageBus };