feat: add job & queue subscription fields

Author: Boris Dorofeev

example/src/demo_queues/fetchMetrics.ts

@@ -33,22 +33,30 @@ metricsQueue.add(
 const metricsWorker = new Worker(
   queueSettings.name,
   async (job) => {
+    // for (let i = 0; i < 5; i++) {
+    //   job.updateProgress(i * 20);
+    //   await new Promise((resolve) => setTimeout(resolve, 1000));
+    // }
+    // console.log(`----> execution of job ${job.id} with data ${JSON.stringify(job.data)}`);
+
     //https://github.com/taskforcesh/bullmq/issues/69
     console.log(new Date().toISOString(), 'Starting name: ' + job.name + ', job: ' + job.id);
     return new Promise((resolve, reject) => {
-      //throw new Error('Ошибка обработки данных...');
       //setTimeout(() => reject('Здесь какая то причина ...'), 0);
       setTimeout(
         () =>
           resolve({
             status: 'job completed',
             result: new Date().toISOString(),
           }),
-        0
+        100000
       );
     });
   },
   {
+    settings: {
+      stalledInterval: 300,
+    },
     prefix,
     connection: createBullConnection('worker'), // BULL_REDIS_URI,
   }

src/composeBull.ts

@@ -34,7 +34,18 @@ import {
   createJobMoveToDelayedFC,
   createQueuePepUpFC,
 } from './mutation';
-import { createOnJobWaitingFC } from './subscriptions';
+import {
+  createOnJobActiveFC,
+  createOnJobCompletedFC,
+  createOnJobDelayedFC,
+  createOnJobFailedFC,
+  createOnJobProgressFC,
+  createOnJobRemovedFC,
+  createOnJobStalledFC,
+  createOnJobWaitingFC,
+  createOnQueuePausedFC,
+  createOnQueueResumedFC,
+} from './subscriptions';
 import { wrapMutationFC, wrapQueueArgs, wrapQueueSubsArgs, composeFC } from './helpers';
 
 interface ComposeBullResult {
@@ -87,11 +98,20 @@ export function composeBull(
     },
   } as ComposeBullResult;
 
-  if (opts?.redisEvents) {
-    data.subscriptionFields = {
-      onJobWaiting: wrapSubscription(createOnJobWaitingFC),
-    };
-  }
+  //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),
+  };
+  //}
 
   return data;
 }

src/helpers/queueEventsListen.ts

@@ -35,7 +35,7 @@ function createAsyncIterator<T = any>(
   prefix: string,
   queueName: string,
   eventName: string
-): AsyncIterator<T> {
+): Required<AsyncIterator<T>> {
   let pullSeries: any = [];
   let pushSeries: any = [];
   let listening = true;
@@ -78,22 +78,20 @@ function createAsyncIterator<T = any>(
     }
   }
 
-  const returnProp = () => {
-    release();
-    return Promise.resolve({ value: undefined, done: true });
-  };
-
   return {
     [Symbol.asyncIterator]() {
       return this;
     },
-    return: returnProp,
+    return() {
+      release();
+      return Promise.resolve({ value: undefined, done: true });
+    },
     next() {
-      return listening ? pullValue() : returnProp();
+      return listening ? pullValue() : this.return();
     },
     throw(error) {
       release();
       return Promise.reject(error);
     },
-  } as AsyncIterator<T>;
+  } as Required<AsyncIterator<T>>;
 }

src/index.ts

@@ -1,24 +1 @@
 export * from './composeBull';
-
-import { getAsyncIterator } from './helpers';
-
-const asyncIter = getAsyncIterator('bull.demo', 'fetch_metrics', 'completed', {
-  typePrefix: 'Prefix',
-});
-
-async function testAsyncIt() {
-  let nn = 3;
-  while (true) {
-    const res = await asyncIter.next();
-    console.log(res);
-
-    nn--;
-
-    if (nn == 0) {
-      await asyncIter.return();
-      break;
-    }
-  }
-}
-
-testAsyncIt();

src/subscriptions/index.ts

@@ -1 +1,10 @@
+export { createOnJobActiveFC } from './onJobActive';
+export { createOnJobCompletedFC } from './onJobCompleted';
+export { createOnJobDelayedFC } from './onJobDelayed';
+export { createOnJobFailedFC } from './onJobFailed';
+export { createOnJobProgressFC } from './onJobProgress';
+export { createOnJobRemovedFC } from './onJobRemoved';
+export { createOnJobStalledFC } from './onJobStalled';
 export { createOnJobWaitingFC } from './onJobWaiting';
+export { createOnQueuePausedFC } from './onQueuePaused';
+export { createOnQueueResumedFC } from './onQueueResumed';

src/subscriptions/onJobActive.ts

@@ -0,0 +1,45 @@
+import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+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 createOnJobActiveFC(
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  return {
+    type: sc.createObjectTC({
+      name: 'OnJobActivePayload',
+      fields: {
+        job: getJobTC(sc, opts),
+        queue: getQueueTC(sc, opts).NonNull,
+        jobId: 'String!',
+        queueName: 'String!',
+        prev: 'String',
+      },
+    }),
+    args: {
+      prefix: {
+        type: 'String!',
+        defaultValue: 'bull',
+      },
+      queueName: 'String!',
+    },
+    resolve: async ({ prefix, queueName, jobId, prev }) => {
+      const queue = getQueue(prefix, queueName, opts);
+      const job = await queue.getJob(jobId);
+      return {
+        job,
+        queue,
+        jobId,
+        queueName,
+        prev,
+      };
+    },
+    subscribe: (_, { prefix, queueName }) => {
+      return getAsyncIterator(prefix, queueName, 'active', opts);
+    },
+  };
+}

src/subscriptions/onJobCompleted.ts

@@ -0,0 +1,43 @@
+import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+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 createOnJobCompletedFC(
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  return {
+    type: sc.createObjectTC({
+      name: 'OnJobCompletedPayload',
+      fields: {
+        job: getJobTC(sc, opts),
+        queue: getQueueTC(sc, opts).NonNull,
+        jobId: 'String!',
+        queueName: 'String!',
+      },
+    }),
+    args: {
+      prefix: {
+        type: 'String!',
+        defaultValue: 'bull',
+      },
+      queueName: 'String!',
+    },
+    resolve: async ({ prefix, queueName, jobId }) => {
+      const queue = getQueue(prefix, queueName, opts);
+      const job = await queue.getJob(jobId);
+      return {
+        job,
+        queue,
+        jobId,
+        queueName,
+      };
+    },
+    subscribe: (_, { prefix, queueName }) => {
+      return getAsyncIterator(prefix, queueName, 'completed', opts);
+    },
+  };
+}

src/subscriptions/onJobDelayed.ts

@@ -0,0 +1,45 @@
+import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+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 createOnJobDelayedFC(
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  return {
+    type: sc.createObjectTC({
+      name: 'OnJobDelayedPayload',
+      fields: {
+        job: getJobTC(sc, opts),
+        queue: getQueueTC(sc, opts).NonNull,
+        jobId: 'String!',
+        queueName: 'String!',
+        delay: 'Int',
+      },
+    }),
+    args: {
+      prefix: {
+        type: 'String!',
+        defaultValue: 'bull',
+      },
+      queueName: 'String!',
+    },
+    resolve: async ({ prefix, queueName, jobId, delay }) => {
+      const queue = getQueue(prefix, queueName, opts);
+      const job = await queue.getJob(jobId);
+      return {
+        job,
+        queue,
+        jobId,
+        queueName,
+        delay: parseInt(delay),
+      };
+    },
+    subscribe: (_, { prefix, queueName }) => {
+      return getAsyncIterator(prefix, queueName, 'delayed', opts);
+    },
+  };
+}

src/subscriptions/onJobFailed.ts

@@ -0,0 +1,43 @@
+import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+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 createOnJobFailedFC(
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  return {
+    type: sc.createObjectTC({
+      name: 'OnJobFailedPayload',
+      fields: {
+        job: getJobTC(sc, opts),
+        queue: getQueueTC(sc, opts).NonNull,
+        jobId: 'String!',
+        queueName: 'String!',
+      },
+    }),
+    args: {
+      prefix: {
+        type: 'String!',
+        defaultValue: 'bull',
+      },
+      queueName: 'String!',
+    },
+    resolve: async ({ prefix, queueName, jobId }) => {
+      const queue = getQueue(prefix, queueName, opts);
+      const job = await queue.getJob(jobId);
+      return {
+        job,
+        queue,
+        jobId,
+        queueName,
+      };
+    },
+    subscribe: (_, { prefix, queueName }) => {
+      return getAsyncIterator(prefix, queueName, 'failed', opts);
+    },
+  };
+}

src/subscriptions/onJobProgress.ts

@@ -0,0 +1,45 @@
+import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+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 createOnJobProgressFC(
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  return {
+    type: sc.createObjectTC({
+      name: 'OnJobProgressPayload',
+      fields: {
+        job: getJobTC(sc, opts),
+        queue: getQueueTC(sc, opts).NonNull,
+        jobId: 'String!',
+        queueName: 'String!',
+        progress: 'Int',
+      },
+    }),
+    args: {
+      prefix: {
+        type: 'String!',
+        defaultValue: 'bull',
+      },
+      queueName: 'String!',
+    },
+    resolve: async ({ prefix, queueName, jobId, data }) => {
+      const queue = getQueue(prefix, queueName, opts);
+      const job = await queue.getJob(jobId);
+      return {
+        job,
+        queue,
+        jobId,
+        queueName,
+        progress: parseInt(data),
+      };
+    },
+    subscribe: (_, { prefix, queueName }) => {
+      return getAsyncIterator(prefix, queueName, 'progress', opts);
+    },
+  };
+}