feat: add redisEvents option which enables Subscriptions for listening events from queues

Author: nodkz

example/src/connectRedis.ts

@@ -5,7 +5,7 @@ const BULL_REDIS_URI = process.env.BULL_REDIS_URI || 'redis://127.0.0.1:6379';
 
 const redisInstances = new Map<string, Redis.Redis>();
 
-export function createBullConnection(type: 'queue' | 'worker' | 'scheduler' | 'custom') {
+export function createBullConnection(type: 'queue' | 'worker' | 'scheduler' | 'events' | 'custom') {
   const existedClient = redisInstances.get(type);
   if (existedClient) {
     return existedClient;

example/src/schema.ts

@@ -11,6 +11,7 @@ const { queryFields, mutationFields, subscriptionFields } = composeBull({
     prefix: 'bull.demo',
   },
   redis: createBullConnection('queue'),
+  redisEvents: createBullConnection('events'),
 });
 
 schemaComposer.Query.addFields({
@@ -21,8 +22,10 @@ schemaComposer.Mutation.addFields({
   ...mutationFields,
 });
 
-schemaComposer.Subscription.addFields({
-  ...subscriptionFields,
-});
+if (subscriptionFields) {
+  schemaComposer.Subscription.addFields({
+    ...subscriptionFields,
+  });
+}
 
 export default schemaComposer.buildSchema();

src/composeBull.ts

@@ -1,4 +1,9 @@
-import { schemaComposer, SchemaComposer } from 'graphql-compose';
+import {
+  schemaComposer,
+  SchemaComposer,
+  ObjectTypeComposer,
+  ObjectTypeComposerFieldConfigAsObjectDefinition,
+} from 'graphql-compose';
 import { Options } from './definitions';
 import { getQueueTC, getJobTC } from './types';
 import {
@@ -29,17 +34,27 @@ import {
   createJobMoveToDelayedFC,
   createQueuePepUpFC,
 } from './mutation';
-import { createJobAddSubFC } from './subscriptions';
+import { createOnJobWaitingFC } from './subscriptions';
 import { wrapMutationFC, wrapQueueArgs, wrapQueueSubsArgs, composeFC } from './helpers';
 
-export function composeBull(opts: Options & { schemaComposer?: SchemaComposer<any> }) {
+interface ComposeBullResult {
+  QueueTC: ObjectTypeComposer;
+  JobTC: ObjectTypeComposer;
+  queryFields: Record<string, ObjectTypeComposerFieldConfigAsObjectDefinition<any, any>>;
+  mutationFields: Record<string, ObjectTypeComposerFieldConfigAsObjectDefinition<any, any>>;
+  subscriptionFields?: Record<string, ObjectTypeComposerFieldConfigAsObjectDefinition<any, any>>;
+}
+
+export function composeBull(
+  opts: Options & { schemaComposer?: SchemaComposer<any> }
+): ComposeBullResult {
   const sc = opts?.schemaComposer || schemaComposer;
 
   const wrapQuery = composeFC(sc, opts)(wrapQueueArgs);
   const wrapMutation = composeFC(sc, opts)(wrapMutationFC, wrapQueueArgs);
   const wrapSubscription = composeFC(sc, opts)(wrapQueueSubsArgs);
 
-  return {
+  const data = {
     QueueTC: getQueueTC(sc, opts),
     JobTC: getJobTC(sc, opts),
     queryFields: {
@@ -70,8 +85,13 @@ export function composeBull(opts: Options & { schemaComposer?: SchemaComposer<an
       jobMoveToDelayed: wrapMutation(createJobMoveToDelayedFC),
       queuePepUp: wrapMutation(createQueuePepUpFC),
     },
-    subscriptionFields: {
-      jobAddSub: wrapSubscription(createJobAddSubFC),
-    },
-  };
+  } as ComposeBullResult;
+
+  if (opts?.redisEvents) {
+    data.subscriptionFields = {
+      onJobWaiting: wrapSubscription(createOnJobWaitingFC),
+    };
+  }
+
+  return data;
 }

src/definitions.ts

@@ -9,4 +9,5 @@ export type Options = {
     prefix?: string;
   };
   redis?: RedisOptions | IORedis.Redis;
+  redisEvents?: RedisOptions | IORedis.Redis;
 };

src/helpers/getBullConnection.ts

@@ -3,7 +3,7 @@ import Redis from 'ioredis';
 
 const connectionMap = new Map<Options['redis'], Redis.Redis>();
 
-export function getBullConnection(opts: Options): Redis.Redis {
+export function getBullConnection(opts?: Options): Redis.Redis {
   const optsRedis = opts?.redis;
   let connection = connectionMap.get(optsRedis);
   if (connection) {

src/helpers/queueEventsListen.ts

@@ -7,13 +7,13 @@ export function getAsyncIterator(
   eventName: string,
   opts: Options
 ) {
-  const queueEvents = getQueueEvents(prefix, queueName, opts);
+  const queueEvents = getQueueEventsSingleton(prefix, queueName, opts);
   return createAsyncIterator(queueEvents, prefix, queueName, eventName);
 }
 
 const queueEventsMap = new Map();
 
-function getQueueEvents(prefix: string, queueName: string, opts: Options): QueueEvents {
+function getQueueEventsSingleton(prefix: string, queueName: string, opts: Options): QueueEvents {
   const fullName = [prefix, queueName].join('.');
 
   if (queueEventsMap.has(fullName)) {
@@ -22,7 +22,7 @@ function getQueueEvents(prefix: string, queueName: string, opts: Options): Queue
 
   const queueEvents = new QueueEvents(queueName, {
     prefix,
-    //connection: new Redis(),
+    connection: opts?.redisEvents,
   });
 
   queueEventsMap.set(fullName, queueEvents);
@@ -36,8 +36,8 @@ function createAsyncIterator<T = any>(
   queueName: string,
   eventName: string
 ): AsyncIterator<T> {
-  const pullSeries: any = [];
-  const pushSeries: any = [];
+  let pullSeries: any = [];
+  let pushSeries: any = [];
   let listening = true;
 
   const pushValue = async (event) => {
@@ -73,12 +73,12 @@ function createAsyncIterator<T = any>(
       for (const resolve of pullSeries) {
         resolve({ value: undefined, done: true });
       }
-      pullSeries.length = 0;
-      pushSeries.length = 0;
+      pullSeries = [];
+      pushSeries = [];
     }
   }
 
-  const returnProp = function () {
+  const returnProp = () => {
     release();
     return Promise.resolve({ value: undefined, done: true });
   };

src/helpers/wrapQueueSubsArgs.ts

@@ -20,10 +20,19 @@ export function wrapQueueSubsArgs(
     if (opts?.queue?.name) predifinedArgs.queueName = opts.queue?.name;
     if (opts?.queue?.prefix) predifinedArgs.prefix = opts.queue?.prefix;
 
-    const subSubscribe = fieldConfig.resolve || (() => ({}));
-    fieldConfig.subscribe = async (source, args, context, info) => {
-      return subSubscribe(source, { ...predifinedArgs, ...args }, context, info);
-    };
+    const subSubscribe = fieldConfig.subscribe;
+    if (subSubscribe) {
+      fieldConfig.subscribe = async (source, args, context, info) => {
+        return subSubscribe(source, { ...predifinedArgs, ...args }, context, info);
+      };
+    }
+
+    const subResolve = fieldConfig.resolve;
+    if (subResolve) {
+      fieldConfig.resolve = async (source, args, context, info) => {
+        return subResolve(source, { ...predifinedArgs, ...args }, context, info);
+      };
+    }
   }
 
   return fieldConfig;

src/subscriptions/index.ts

@@ -1 +1 @@
-export { createJobAddSubFC } from './jobAddSub';
+export { createOnJobWaitingFC } from './onJobWaiting';

src/subscriptions/jobAddSub.ts renamed to src/subscriptions/onJobWaiting.ts

@@ -3,13 +3,22 @@ import { getJobTC } from '../types/job/Job';
 import { getQueue } from '../helpers';
 import { Options } from '../definitions';
 import { getAsyncIterator } from '../helpers';
+import { getQueueTC } from '../types/queue/Queue';
 
-export function createJobAddSubFC(
+export function createOnJobWaitingFC(
   sc: SchemaComposer<any>,
   opts: Options
 ): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
   return {
-    type: getJobTC(sc, opts),
+    type: sc.createObjectTC({
+      name: 'OnWaitingPayload',
+      fields: {
+        job: getJobTC(sc, opts),
+        queue: getQueueTC(sc, opts).NonNull,
+        jobId: 'String!',
+        queueName: 'String!',
+      },
+    }),
     args: {
       prefix: {
         type: 'String!',
@@ -19,10 +28,13 @@ export function createJobAddSubFC(
     },
     resolve: async ({ prefix, queueName, jobId }) => {
       const queue = getQueue(prefix, queueName, opts);
-      if (!queue) return null;
       const job = await queue.getJob(jobId);
-      if (!job) return null;
-      return job;
+      return {
+        job,
+        queue,
+        jobId,
+        queueName,
+      };
     },
     subscribe: (_, { prefix, queueName }) => {
       return getAsyncIterator(prefix, queueName, 'waiting', opts);