fix: single connection for queue-events

Author: Boris Dorofeev

src/composeBull.ts

@@ -65,7 +65,7 @@ export function composeBull(
   const wrapMutation = composeFC(sc, opts)(wrapMutationFC, wrapQueueArgs);
   const wrapSubscription = composeFC(sc, opts)(wrapQueueSubsArgs);
 
-  const data = {
+  const data: ComposeBullResult = {
     QueueTC: getQueueTC(sc, opts),
     JobTC: getJobTC(sc, opts),
     queryFields: {
@@ -96,22 +96,19 @@ export function composeBull(
       jobMoveToDelayed: wrapMutation(createJobMoveToDelayedFC),
       queuePepUp: wrapMutation(createQueuePepUpFC),
     },
-  } as ComposeBullResult;
-
-  //if (opts?.redisEvents) {
-  data.subscriptionFields = {
-    onJobActive: wrapSubscription(createOnJobActiveFC),
-    onJobCompleted: wrapSubscription(createOnJobCompletedFC),
-    onJobDelayed: wrapSubscription(createOnJobDelayedFC),
-    onJobFailed: wrapSubscription(createOnJobFailedFC),
-    onJobProgress: wrapSubscription(createOnJobProgressFC),
-    onJobRemoved: wrapSubscription(createOnJobRemovedFC),
-    onJobStalled: wrapSubscription(createOnJobStalledFC),
-    onJobWaiting: wrapSubscription(createOnJobWaitingFC),
-    onQueuePaused: wrapSubscription(createOnQueuePausedFC),
-    onQueueResumed: wrapSubscription(createOnQueueResumedFC),
+    subscriptionFields: {
+      onJobActive: wrapSubscription(createOnJobActiveFC),
+      onJobCompleted: wrapSubscription(createOnJobCompletedFC),
+      onJobDelayed: wrapSubscription(createOnJobDelayedFC),
+      onJobFailed: wrapSubscription(createOnJobFailedFC),
+      onJobProgress: wrapSubscription(createOnJobProgressFC),
+      onJobRemoved: wrapSubscription(createOnJobRemovedFC),
+      onJobStalled: wrapSubscription(createOnJobStalledFC),
+      onJobWaiting: wrapSubscription(createOnJobWaitingFC),
+      onQueuePaused: wrapSubscription(createOnQueuePausedFC),
+      onQueueResumed: wrapSubscription(createOnQueueResumedFC),
+    },
   };
-  //}
 
   return data;
 }

src/helpers/queueEventsListen.ts

@@ -1,5 +1,6 @@
 import { QueueEvents } from 'bullmq';
 import { Options } from '../definitions';
+import { getBullConnection } from './getBullConnection';
 
 export function getAsyncIterator(
   prefix: string,
@@ -22,7 +23,7 @@ function getQueueEventsSingleton(prefix: string, queueName: string, opts: Option
 
   const queueEvents = new QueueEvents(queueName, {
     prefix,
-    connection: opts?.redisEvents,
+    connection: getBullConnection(opts),
   });
 
   queueEventsMap.set(fullName, queueEvents);