refactor: event-bus inherits EventEmitter (#235)

Author: Uzlopak

lib/event-bus.js

@@ -6,64 +6,60 @@ import { EventEmitter } from 'node:events'
  * allowing payloads to be received by one instance of Smee and sent by
  * others. This allows Smee to be multi-instanced!
  */
-export default class EventBus {
+export default class EventBus extends EventEmitter {
+  #namespace = ''
+
+  #connection = ''
+
+  /** @type {Redis} */
+  #sub
+
+  /** @type {Redis} */
+  #pub
+
+  #connected = false
+
   constructor ({ logger = console } = {}) {
-    if (!(this instanceof EventBus)) {
-      return new EventBus()
-    }
-    this.events = new EventEmitter()
+    super()
 
     // If Redis isn't enabled, don't try to connect
     if (!process.env.REDIS_URL) {
       logger.warn('Redis not enabled; events will not be shared between instances')
       return
     }
 
-    this.opts = {
-      connection: process.env.REDIS_URL,
-      namespace: 'global'
-    }
+    this.#namespace = process.env.REDIS_NAMESPACE || 'global'
+
+    this.#connection = process.env.REDIS_URL || ''
 
     // Need two Redis clients; one cannot subscribe and publish.
-    this.sub = new Redis(this.opts.connection)
-    this.pub = new Redis(this.opts.connection)
+    this.#sub = new Redis(this.#connection)
+    this.#pub = new Redis(this.#connection)
 
     // Subscribe to the Redis event channel
-    this.sub.subscribe(this.opts.namespace)
-
-    logger.info(`Redis enabled; events will be shared between instances using ${this.opts.namespace} as the namespace`)
+    this.#sub.subscribe(this.#namespace)
 
+    
     // When we get a message, parse it and
     // throw it over to the EventEmitter.
-    this.sub.on('message', (_, message) => {
+    this.#sub.on('message', (_, message) => {
       const channel = message.slice(0, message.indexOf(':'))
-      if (this.events.listenerCount(channel)) {
-        return this.events.emit(channel, message.slice(channel.length + 1))
+      if (this.listenerCount(channel)) {
+        return super.emit(channel, message.slice(channel.length + 1))
       }
     })
+    
+    this.#connected = true
+    logger.info(`Redis enabled; events will be shared between instances using '${this.#namespace}' as the namespace`)
   }
 
   /**
- * Emit an event to this machine's in-memory EventEmitter
- * @param {object} opts
- * @param {string} opts.channel - Channel name
- * @param {any} opts.payload
- */
-  emitLocalEvent (opts) {
-    return this.events.emit(opts.channel, opts.payload)
-  }
-
-  /**
-   * Emit an event to the Redis bus, which will tell every subscriber about it
-   * @param {object} opts
-   * @param {string} opts.channel - Channel name
-   * @param {any} opts.payload
+   * @param {string} channelId
+   * @param {string} payload
    */
-  emitEvent (opts) {
-    if (process.env.REDIS_URL) {
-      return this.pub.publish(this.opts.namespace, opts.channel + ':' + JSON.stringify(opts.payload))
-    }
-    // Only emit local events if Redis isn't configured
-    return this.emitLocalEvent(opts)
+  async emit (channelId, payload) {
+    return this.#connected
+      ? this.#pub.publish(this.#namespace, channelId + ':' + JSON.stringify(payload)) !== 0
+      : super.emit(channelId, payload)
   }
 }

lib/server.js

@@ -227,24 +227,24 @@ const [
         })
 
       function close () {
-        bus.events.removeListener(channel, reply.sse)
+        bus.removeListener(channel, reply.sse)
         keepAlive.stop(reply)
-        fastify.log.info('Client disconnected', channel, bus.events.listenerCount(channel))
+        fastify.log.info('Client disconnected', channel, bus.listenerCount(channel))
       }
 
       // Listen for events on this channel
-      bus.events.on(channel, reply.sse.bind(reply))
+      bus.on(channel, reply.sse.bind(reply))
 
       // Clean up when the client disconnects
       reply.raw.on('close', close)
 
       reply.sseReady()
 
-      fastify.log.info('Client connected to sse', channel, bus.events.listenerCount(channel))
+      fastify.log.info('Client connected to sse', channel, bus.listenerCount(channel))
       return
     }
 
-    fastify.log.info('Client connected to web', channel, bus.events.listenerCount(channel))
+    fastify.log.info('Client connected to web', channel, bus.listenerCount(channel))
     return reply
       .status(200)
       .header('Content-Type', 'text/html; charset=utf-8')
@@ -271,14 +271,13 @@ const [
     }
   }, async (req, reply) => {
     // Emit an event to the Redis bus
-    await bus.emitEvent({
-      channel: req.params.channel,
-      payload: {
+    await bus.emit(
+      req.params.channel,
+      {
         ...req.headers,
         body: req.body,
         query: req.query,
         timestamp: Date.now()
-      }
     })
 
     return reply.status(200).send()
@@ -314,7 +313,7 @@ const [
       }
     }, async (req, reply) => {
       // Emit an event to the Redis bus
-      await bus.emitEvent({
+      await bus.emit({
         channel: req.params.channel,
         payload: req.body
       })