refactor: bus

Author: Julien-R44

packages/bentocache/benchmarks/encoders.ts

@@ -1,10 +1,10 @@
 import lodash from '@poppinss/utils/lodash'
 import { getActiveTest } from '@japa/runner'
+import { MemoryTransport } from '@rlanz/bus/drivers/memory'
 
 import { Cache } from '../src/cache/cache.js'
 import { RedisDriver } from '../src/drivers/redis.js'
 import { MemoryDriver } from '../src/drivers/memory.js'
-import { MemoryBus } from '../src/bus/drivers/memory_bus.js'
 import type { CacheStackDrivers } from '../src/types/main.js'
 import { CacheStack } from '../src/cache/stack/cache_stack.js'
 import { BentoCacheOptions } from '../src/bento_cache_options.js'
@@ -80,7 +80,7 @@ export class CacheFactory {
   withL1L2Config() {
     this.#parameters.l1Driver ??= new MemoryDriver({ maxSize: 100, prefix: 'test' })
     this.#parameters.l2Driver ??= new RedisDriver({ connection: { host: '127.0.0.1', port: 6379 } })
-    this.#parameters.busDriver ??= new MemoryBus()
+    this.#parameters.busDriver ??= new MemoryTransport()
 
     return this
   }

packages/bentocache/factories/cache_factory.ts

@@ -69,6 +69,7 @@
   "dependencies": {
     "@paralleldrive/cuid2": "^2.2.2",
     "@poppinss/utils": "^6.7.3",
+    "@rlanz/bus": "^0.2.0",
     "async-mutex": "^0.5.0",
     "chunkify": "^5.0.0",
     "hexoid": "^1.0.0",

packages/bentocache/package.json

@@ -1,11 +1,12 @@
+import { Bus as RlanzBus } from '@rlanz/bus'
 import { createId } from '@paralleldrive/cuid2'
+import type { Transport } from '@rlanz/bus/types/main'
 
-import { RetryQueue } from './retry_queue.js'
 import { CacheBusMessageType } from '../types/bus.js'
 import type { LocalCache } from '../cache/facades/local_cache.js'
 import { BusMessageReceived } from '../events/bus/bus_message_received.js'
 import { BusMessagePublished } from '../events/bus/bus_message_published.js'
-import type { BusDriver, BusOptions, CacheBusMessage, Emitter, Logger } from '../types/main.js'
+import type { BusOptions, CacheBusMessage, Emitter, Logger } from '../types/main.js'
 
 /**
  * The bus is used to notify other processes about cache changes.
@@ -17,94 +18,51 @@ import type { BusDriver, BusOptions, CacheBusMessage, Emitter, Logger } from '..
  * local cache accordingly.
  */
 export class Bus {
-  /**
-   * The underlying bus driver
-   */
-  #driver: BusDriver
-
   /**
    * The local cache that will be updated when a message is received
    */
   #cache?: LocalCache
 
-  /**
-   * The logger to use
-   */
-  #logger: Logger
-
-  /**
-   * Emitter
-   */
-  #emitter: Emitter
-
   /**
    * A unique identifier for this bus instance
    * that is used to prevent the bus from
    * emitting events to itself
    */
   #busId = createId()
-
-  /**
-   * The channel name to use
-   */
+  #bus: RlanzBus
+  #logger: Logger
+  #emitter: Emitter
   #channelName = 'bentocache.notifications'
 
-  /**
-   * The error retry queue holds messages that failed to be sent
-   */
-  #errorRetryQueue = new RetryQueue()
-
   constructor(
-    driver: BusDriver,
+    driver: Transport,
     cache: LocalCache,
     logger: Logger,
     emitter: Emitter,
     options: BusOptions = {},
   ) {
-    this.#driver = driver
     this.#cache = cache
     this.#emitter = emitter
     this.#logger = logger.child({ context: 'bentocache.bus' })
-    this.#errorRetryQueue = new RetryQueue(options.retryQueue?.enabled, options.retryQueue?.maxSize)
 
-    driver
-      .setId(this.#busId)
-      .setLogger(this.#logger)
-      .onReconnect(() => this.#onReconnect())
-  }
+    this.#bus = new RlanzBus(driver, {
+      retryQueue: { ...options.retryQueue, removeDuplicates: true, retryInterval: false },
+    })
 
-  /**
-   * Process the error retry queue
-   */
-  async #processErrorRetryQueue() {
-    this.#logger.debug(
-      `starting error retry queue processing with ${this.#errorRetryQueue.size()} messages`,
+    this.#bus.subscribe<CacheBusMessage & { [key: string]: any }>(
+      this.#channelName,
+      this.#onMessage.bind(this),
     )
-
-    await this.#errorRetryQueue.process(async (message) => {
-      await this.publish(message)
-      return true
-    })
   }
 
   /**
    * When a message is received through the bus.
    * This is where we update the local cache.
    */
   async #onMessage(message: CacheBusMessage) {
-    /**
-     * Since we received a message from the bus, we assume that
-     * the Bus is working. So we can try process the error retry queue if
-     * there are any messages in it.
-     */
-    await this.#processErrorRetryQueue()
-
     this.#logger.trace({ keys: message.keys, type: message.type }, 'received message from bus')
     this.#emitter.emit('bus:message:received', new BusMessageReceived(message))
 
-    /**
-     * Process the message
-     */
     if (message.type === CacheBusMessageType.Delete) {
       for (const key of message.keys) this.#cache?.delete(key)
     }
@@ -118,52 +76,19 @@ export class Bus {
     }
   }
 
-  /**
-   * When the bus driver reconnects after a disconnection
-   */
-  async #onReconnect() {
-    this.#logger.debug('bus driver reconnected')
-    await this.#processErrorRetryQueue()
-  }
-
-  /**
-   * Subscribe to the bus channel
-   */
-  async subscribe() {
-    this.#driver.subscribe(this.#channelName, this.#onMessage.bind(this))
-  }
-
   /**
    * Publish a message to the bus channel
    *
    * @returns true if the message was published, false if not
    */
   async publish(message: Omit<CacheBusMessage, 'busId'>): Promise<boolean> {
-    const fullMessage = { ...message, busId: this.#busId }
-
     try {
-      this.#logger.trace({ keys: message.keys, type: message.type }, 'publishing message to bus')
-
-      /**
-       * Publish the message to the bus using the underlying driver
-       */
-      await this.#driver.publish(this.#channelName, fullMessage)
-
-      /**
-       * Emit the bus:message:published event
-       */
+      const fullMessage = { ...message, busId: this.#busId }
+      await this.#bus.publish(this.#channelName, message)
       this.#emitter.emit('bus:message:published', new BusMessagePublished(fullMessage))
       return true
     } catch (error) {
-      this.#logger.error(error, 'failed to publish message to bus')
-
-      /**
-       * Add to the error retry queue
-       */
-      const wasAdded = this.#errorRetryQueue.enqueue(fullMessage)
-      if (!wasAdded) return false
-
-      this.#logger.debug(message, 'added message to error retry queue')
+      this.#logger.error({ error }, 'failed to publish message to bus')
       return false
     }
   }
@@ -172,6 +97,6 @@ export class Bus {
    * Disconnect the bus
    */
   async disconnect(): Promise<void> {
-    this.#driver.disconnect()
+    await this.#bus.disconnect()
   }
 }