fix: support for cluster with bus (#98)

Author: Julien-R44

.changeset/fine-rabbits-glow.md

@@ -0,0 +1,5 @@
+---
+'bentocache': patch
+---
+
+Add support for Cluster with bus

.gitignore

@@ -3,8 +3,7 @@ build
 coverage
 poc
 doc.md
-.env*
-!.env.*.example
+.env
 cache.sqlite3
 .todo.md
 

docs/vscode_grammars/dotenv.tm_language.json

@@ -9,7 +9,6 @@
     ".env.dev",
     ".env.test",
     ".env.testing",
-    ".env.testing.example",
     ".env.production",
     ".env.prod"
   ],
@@ -147,4 +146,4 @@
       ]
     }
   }
-}
+}

packages/bentocache/.env.testing.example

@@ -1,14 +1,12 @@
-import { config } from 'dotenv'
-import { resolve } from 'node:path'
+import 'dotenv/config'
+
 import { assert } from '@japa/assert'
 import { fileSystem } from '@japa/file-system'
 import { expectTypeOf } from '@japa/expect-type'
 import { processCLIArgs, configure, run } from '@japa/runner'
 
 import { BASE_URL } from '../tests/helpers/index.js'
 
-config({ path: resolve(process.cwd(), '.env.testing') })
-
 /*
 |--------------------------------------------------------------------------
 | Configure tests

packages/bentocache/bin/test.ts

@@ -91,7 +91,7 @@
     }
   },
   "dependencies": {
-    "@boringnode/bus": "^0.7.1",
+    "@boringnode/bus": "^0.9.0",
     "@julr/utils": "^1.9.0",
     "@poppinss/exception": "^1.2.1",
     "async-mutex": "^0.5.0",

packages/bentocache/package.json

@@ -23,13 +23,24 @@ export function redisDriver(options: RedisConfig): CreateDriverResult<RedisDrive
 /**
  * Create a new bus redis driver. It leverages the Pub/sub capabilities of Redis
  * to sending messages between your different processes.
+ *
+ * You can pass either connection options or an existing Redis/Cluster instance.
  */
 export function redisBusDriver(
-  options: { connection: IoRedisOptions } & BusOptions,
+  options: { connection: IoRedisOptions | IoRedis | IoRedisCluster } & BusOptions,
 ): CreateBusDriverResult {
   return {
     options,
     factory: () => {
+      /**
+       * If an existing Redis or Cluster instance is passed, use it directly
+       */
+      if (options.connection instanceof IoRedis || options.connection instanceof IoRedisCluster) {
+        return new RedisTransport(options.connection, new BinaryEncoder(), {
+          useMessageBuffer: true,
+        })
+      }
+
       return new RedisTransport(
         { ...options.connection, useMessageBuffer: true } as RedisTransportConfig,
         new BinaryEncoder(),

packages/bentocache/src/drivers/redis.ts

@@ -1,6 +1,7 @@
 import { test } from '@japa/runner'
 import { sleep } from '@julr/utils/misc'
 import { MemoryTransport } from '@boringnode/bus/transports/memory'
+import { Redis as IoRedis, Cluster as IoRedisCluster } from 'ioredis'
 
 import { ChaosBus } from '../helpers/chaos/chaos_bus.js'
 import { ChaosCache } from '../helpers/chaos/chaos_cache.js'
@@ -411,4 +412,70 @@ test.group('Bus synchronization', () => {
 
     await sleep(200)
   }).waitForDone()
+
+  test('works with an existing Redis instance', async ({ assert }, done) => {
+    const redis = new IoRedis(REDIS_CREDENTIALS)
+
+    const bus1 = redisBusDriver({ connection: redis })
+      .factory(null as any)
+      .setId('foo')
+
+    const bus2 = redisBusDriver({ connection: redis })
+      .factory(null as any)
+      .setId('bar')
+
+    const data = {
+      keys: ['foo'],
+      type: CacheBusMessageType.Set,
+    }
+
+    bus1.subscribe('foo', (message: any) => {
+      assert.deepInclude(message, data)
+      done()
+    })
+
+    await sleep(200)
+
+    await bus2.publish('foo', data)
+
+    await bus1.disconnect()
+    await bus2.disconnect()
+    await redis.quit()
+
+    await sleep(200)
+  }).waitForDone()
+
+  test('works with an existing Cluster instance', async ({ assert }, done) => {
+    const cluster = new IoRedisCluster([{ host: '127.0.0.1', port: 7000 }])
+
+    const bus1 = redisBusDriver({ connection: cluster })
+      .factory(null as any)
+      .setId('foo')
+
+    const bus2 = redisBusDriver({ connection: cluster })
+      .factory(null as any)
+      .setId('bar')
+
+    const data = {
+      keys: ['foo'],
+      type: CacheBusMessageType.Set,
+    }
+
+    bus1.subscribe('foo', (message: any) => {
+      assert.deepInclude(message, data)
+      done()
+    })
+
+    await sleep(200)
+
+    await bus2.publish('foo', data)
+
+    await bus1.disconnect()
+    await bus2.disconnect()
+    await cluster.quit()
+
+    await sleep(200)
+  })
+    .waitForDone()
+    .skip(!!process.env.CI, 'Skipping cluster test on CI')
 })

packages/bentocache/tests/bus/bus.spec.ts

@@ -6,7 +6,7 @@ import { registerCacheDriverTestSuite } from '../helpers/driver_test_suite.js'
 
 const dynamoClient = new DynamoDBClient({
   region: 'eu-west-3',
-  endpoint: process.env.DYNAMODB_ENDPOINT,
+  endpoint: 'http://localhost:8000',
   credentials: { accessKeyId: 'foo', secretAccessKey: 'foo' },
 })
 

packages/bentocache/tests/drivers/dynamodb.spec.ts

@@ -3,7 +3,6 @@ import { test } from '@japa/runner'
 import { sleep } from '@julr/utils/misc'
 
 import { createKnexStore } from './helpers.js'
-import { DatabaseType, getDbConfig } from '../../helpers/db_config.js'
 import { registerCacheDriverTestSuite } from '../../helpers/driver_test_suite.js'
 
 test.group('Knex | MySQL driver', (group) => {
@@ -12,13 +11,20 @@ test.group('Knex | MySQL driver', (group) => {
     group,
     supportsMilliseconds: false,
     createDriver: (options) => {
-      const db = knex(getDbConfig(DatabaseType.MYSQL))
+      const db = knex({
+        client: 'mysql2',
+        connection: { user: 'root', password: 'root', database: 'mysql', port: 3306 },
+      })
+
       return createKnexStore({ connection: db, prefix: 'japa', ...options })
     },
   })
 
   test('should not throw error when disconnecting immediately', async () => {
-    const db = knex(getDbConfig(DatabaseType.MYSQL))
+    const db = knex({
+      client: 'mysql2',
+      connection: { user: 'root', password: 'root', database: 'mysql', port: 3306 },
+    })
 
     const store = createKnexStore({ connection: db, prefix: 'japa' })