feat: add async iterator

Author: Boris Dorofeev

example/src/schema.ts

@@ -2,7 +2,7 @@ import { composeBull } from '../../src';
 import { schemaComposer } from 'graphql-compose';
 import { createBullConnection } from './connectRedis';
 
-const { queryFields, mutationFields } = composeBull({
+const { queryFields, mutationFields, subscriptionFields } = composeBull({
   schemaComposer,
   typePrefix: 'Prefix',
   jobDataTC: `type MyJobData { fieldA: String! fieldB: String}`,
@@ -21,4 +21,8 @@ schemaComposer.Mutation.addFields({
   ...mutationFields,
 });
 
+schemaComposer.Subscription.addFields({
+  ...subscriptionFields,
+});
+
 export default schemaComposer.buildSchema();

package.json

@@ -25,8 +25,8 @@
   },
   "homepage": "https://github.com/graphql-compose/graphql-compose-bullmq",
   "peerDependencies": {
-    "graphql-compose": "^7.15.0",
-    "bullmq": "^1.8.11"
+    "bullmq": "^1.8.11",
+    "graphql-compose": "^7.15.0"
   },
   "devDependencies": {
     "@types/express": "^4.17.3",

src/composeBull.ts

@@ -29,13 +29,15 @@ import {
   createJobMoveToDelayedFC,
   createQueuePepUpFC,
 } from './mutation';
-import { wrapMutationFC, wrapQueueArgs, composeFC } from './helpers';
+import { createJobAddSubFC } from './subscriptions';
+import { wrapMutationFC, wrapQueueArgs, wrapQueueSubsArgs, composeFC } from './helpers';
 
 export function composeBull(opts: Options & { schemaComposer?: SchemaComposer<any> }) {
   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 {
     QueueTC: getQueueTC(sc, opts),
@@ -68,5 +70,8 @@ export function composeBull(opts: Options & { schemaComposer?: SchemaComposer<an
       jobMoveToDelayed: wrapMutation(createJobMoveToDelayedFC),
       queuePepUp: wrapMutation(createQueuePepUpFC),
     },
+    subscriptionFields: {
+      jobAddSub: wrapSubscription(createJobAddSubFC),
+    },
   };
 }

src/helpers/index.ts

@@ -4,7 +4,9 @@ export * from './MutationError';
 export * from './normalizePrefixGlob';
 export * from './wrapMutationFC';
 export * from './wrapQueueArgs';
+export * from './wrapQueueSubsArgs';
 export * from './composeFC';
 export * from './deleteKeys';
 export * from './fixDelayStream';
 export * from './memoryUsage';
+export * from './queueEventsListen';

src/helpers/queueEventsListen.ts

@@ -0,0 +1,99 @@
+import { QueueEvents } from 'bullmq';
+import { Options } from '../definitions';
+
+export function getAsyncIterator(
+  prefix: string,
+  queueName: string,
+  eventName: string,
+  opts: Options
+) {
+  const queueEvents = getQueueEvents(prefix, queueName, opts);
+  return createAsyncIterator(queueEvents, prefix, queueName, eventName);
+}
+
+const queueEventsMap = new Map();
+
+function getQueueEvents(prefix: string, queueName: string, opts: Options): QueueEvents {
+  const fullName = [prefix, queueName].join('.');
+
+  if (queueEventsMap.has(fullName)) {
+    return queueEventsMap.get(fullName);
+  }
+
+  const queueEvents = new QueueEvents(queueName, {
+    prefix,
+    //connection: new Redis(),
+  });
+
+  queueEventsMap.set(fullName, queueEvents);
+
+  return queueEvents;
+}
+
+function createAsyncIterator<T = any>(
+  queueEvents: QueueEvents,
+  prefix: string,
+  queueName: string,
+  eventName: string
+): AsyncIterator<T> {
+  const pullSeries: any = [];
+  const pushSeries: any = [];
+  let listening = true;
+
+  const pushValue = async (event) => {
+    if (pullSeries.length !== 0) {
+      const resolver = pullSeries.shift();
+      resolver(event);
+    } else {
+      pushSeries.push(event);
+    }
+  };
+
+  const pullValue = () => {
+    return new Promise((resolve) => {
+      if (pushSeries.length !== 0) {
+        const value = pushSeries.shift();
+        resolve({ value, done: false });
+      } else {
+        pullSeries.push(resolve);
+      }
+    });
+  };
+
+  const handler = (event) => {
+    pushValue({ prefix, queueName, ...event });
+  };
+
+  queueEvents.on(eventName, handler);
+
+  function release() {
+    if (listening) {
+      listening = false;
+      queueEvents.removeListener(eventName, handler);
+      for (const resolve of pullSeries) {
+        resolve({ value: undefined, done: true });
+      }
+      pullSeries.length = 0;
+      pushSeries.length = 0;
+    }
+  }
+
+  const returnProp = function () {
+    release();
+    return Promise.resolve({ value: undefined, done: true });
+  };
+
+  return {
+    [Symbol.asyncIterator]() {
+      return this;
+    },
+    return: returnProp,
+    next() {
+      return listening ? pullValue() : returnProp();
+    },
+    throw(error) {
+      release();
+      return Promise.reject(error);
+    },
+  } as AsyncIterator<T>;
+}

src/helpers/wrapQueueSubsArgs.ts

@@ -0,0 +1,30 @@
+import { Options } from '../definitions';
+import { ObjectTypeComposerFieldConfigAsObjectDefinition, SchemaComposer } from 'graphql-compose';
+
+export function wrapQueueSubsArgs(
+  fieldConfig: ObjectTypeComposerFieldConfigAsObjectDefinition<any, any>,
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  // remove args if they provided via config
+  if (opts?.queue?.name && fieldConfig.args?.queueName) {
+    delete fieldConfig.args.queueName;
+  }
+  if (opts?.queue?.prefix && fieldConfig.args?.prefix) {
+    delete fieldConfig.args.prefix;
+  }
+
+  // pass config props to sub resolve issue
+  if (opts?.queue?.name || opts?.queue?.prefix) {
+    const predifinedArgs = {} as Record<string, any>;
+    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);
+    };
+  }
+
+  return fieldConfig;
+}

src/index.ts

@@ -1 +1,24 @@
 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

@@ -0,0 +1 @@
+export { createJobAddSubFC } from './jobAddSub';

src/subscriptions/jobAddSub.ts

@@ -0,0 +1,31 @@
+import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+import { getJobTC } from '../types/job/Job';
+import { getQueue } from '../helpers';
+import { Options } from '../definitions';
+import { getAsyncIterator } from '../helpers';
+
+export function createJobAddSubFC(
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  return {
+    type: getJobTC(sc, opts),
+    args: {
+      prefix: {
+        type: 'String!',
+        defaultValue: 'bull',
+      },
+      queueName: 'String!',
+    },
+    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;
+    },
+    subscribe: (_, { prefix, queueName }) => {
+      return getAsyncIterator(prefix, queueName, 'waiting', opts);
+    },
+  };
+}