ci: add support for testing on dragonflydb (#2210)

.github/workflows/test.yml

@@ -58,7 +58,7 @@ jobs:
             exit 1
           fi
 
-  compatibility:
+  node-redis:
     runs-on: ubuntu-latest
 
     name: testing node@${{ matrix.node-version }}, redis@${{ matrix.redis-version }}
@@ -87,6 +87,35 @@ jobs:
       - run: yarn build
       - run: yarn test
 
+  node-dragonflydb:
+    runs-on: ubuntu-latest
+
+    env:
+      node-version: lts/*
+
+    name: testing node@lts/*, dragonflydb@latest
+
+    services:
+      dragonflydb:
+        image: docker.dragonflydb.io/dragonflydb/dragonfly
+        env:
+          DFLY_cluster_mode: emulated
+          DFLY_lock_on_hashtags: true
+        ports:
+          - 6379:6379
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3 # v3
+      - name: Use Node.js ${{ env.node-version }}
+        uses: actions/setup-node@v3 # v3
+        with:
+          node-version: lts/*
+          cache: 'yarn'
+      - run: yarn install --ignore-engines --frozen-lockfile --non-interactive
+      - run: yarn build
+      - run: BULLMQ_TEST_PREFIX={b} yarn test
+
   python:
     runs-on: ubuntu-latest
 

.mocharc.js

@@ -3,4 +3,5 @@ module.exports = {
   file: ['./mocha.setup.ts'],
   spec: ['./tests/test_*.ts'],
   timeout: 4000,
+  'trace-warnings': true,
 };

src/classes/queue-getters.ts

@@ -221,7 +221,8 @@ export class QueueGetters<
   }
 
   /**
-   * Returns the jobs that are in the "waiting" status.
+   * Returns the jobs that are in the "waiting-children" status.
+   * I.E. parent jobs that have at least one child that has not completed yet.
    * @param start - zero based index from where to start returning jobs.
    * @param end - zero based index where to stop returning jobs.
    */

src/classes/redis-connection.ts

@@ -222,11 +222,15 @@ export class RedisConnection extends EventEmitter {
     return this._client;
   }
 
-  async disconnect(): Promise<void> {
+  async disconnect(wait = true): Promise<void> {
     const client = await this.client;
     if (client.status !== 'end') {
       let _resolve, _reject;
 
+      if (!wait) {
+        return client.disconnect();
+      }
+
       const disconnecting = new Promise<void>((resolve, reject) => {
         client.once('end', resolve);
         client.once('error', reject);

src/classes/worker.ts

@@ -862,7 +862,8 @@ export class Worker<
     // Force reconnection of blocking connection to abort blocking redis call immediately.
     //
     if (this.waiting) {
-      await this.blockingConnection.disconnect();
+      // If we are not going to reconnect, we will not wait for the disconnection.
+      await this.blockingConnection.disconnect(reconnect);
     } else {
       reconnect = false;
     }

src/utils.ts

@@ -81,7 +81,7 @@ export function isRedisCluster(obj: unknown): obj is Cluster {
 export async function removeAllQueueData(
   client: RedisClient,
   queueName: string,
-  prefix = 'bull',
+  prefix = process.env.BULLMQ_TEST_PREFIX || 'bull',
 ): Promise<void | boolean> {
   if (client instanceof Cluster) {
     // todo compat with cluster ?

tests/test_bulk.ts

@@ -5,14 +5,16 @@ import { v4 } from 'uuid';
 import { Queue, Worker, Job } from '../src/classes';
 import { removeAllQueueData } from '../src/utils';
 
+const prefix = process.env.BULLMQ_TEST_PREFIX || 'bull';
+
 describe('bulk jobs', () => {
   let queue: Queue;
   let queueName: string;
   const connection = { host: 'localhost' };
 
   beforeEach(async function () {
     queueName = `test-${v4()}`;
-    queue = new Queue(queueName, { connection });
+    queue = new Queue(queueName, { connection, prefix });
   });
 
   afterEach(async function () {
@@ -35,7 +37,7 @@ describe('bulk jobs', () => {
           }
         }),
     );
-    const worker = new Worker(queueName, processor, { connection });
+    const worker = new Worker(queueName, processor, { connection, prefix });
     await worker.waitUntilReady();
 
     const jobs = await queue.addBulk([
@@ -56,10 +58,13 @@ describe('bulk jobs', () => {
   it('should allow to pass parent option', async () => {
     const name = 'test';
     const parentQueueName = `parent-queue-${v4()}`;
-    const parentQueue = new Queue(parentQueueName, { connection });
+    const parentQueue = new Queue(parentQueueName, { connection, prefix });
 
-    const parentWorker = new Worker(parentQueueName, null, { connection });
-    const childrenWorker = new Worker(queueName, null, { connection });
+    const parentWorker = new Worker(parentQueueName, null, {
+      connection,
+      prefix,
+    });
+    const childrenWorker = new Worker(queueName, null, { connection, prefix });
     await parentWorker.waitUntilReady();
     await childrenWorker.waitUntilReady();
 
@@ -71,7 +76,7 @@ describe('bulk jobs', () => {
         opts: {
           parent: {
             id: parent.id,
-            queue: `bull:${parentQueueName}`,
+            queue: `${prefix}:${parentQueueName}`,
           },
         },
       },
@@ -81,7 +86,7 @@ describe('bulk jobs', () => {
         opts: {
           parent: {
             id: parent.id,
-            queue: `bull:${parentQueueName}`,
+            queue: `${prefix}:${parentQueueName}`,
           },
         },
       },
@@ -120,7 +125,7 @@ describe('bulk jobs', () => {
           }
         }),
     );
-    const worker = new Worker(queueName, processor, { connection });
+    const worker = new Worker(queueName, processor, { connection, prefix });
     await worker.waitUntilReady();
 
     const jobs = await queue.addBulk([