chore: fix AsyncIterator return value

Author: nodkz

example/src/demo_queues/fetchMetrics.ts

@@ -11,6 +11,7 @@ const prefix = queueSettings.prefix;
 const metricsScheduler = new QueueScheduler(queueSettings.name, {
   prefix,
   connection: createBullConnection('scheduler'), // BULL_REDIS_URI,
+  stalledInterval: 900000,
 });
 
 export const metricsQueue = new Queue(queueSettings.name, {
@@ -24,39 +25,33 @@ metricsQueue.add(
   { repeat: { cron: '*/1 * * * *' } }
 );
 
-metricsQueue.add(
-  'fetch_metrics_every_100000',
-  { field1: 'asdasdadas' },
-  { repeat: { every: 100000 } }
-);
+metricsQueue.add('fetch_metrics_every_5000', { field1: 'asdasdadas' }, { repeat: { every: 5000 } });
 
 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)}`);
+    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) => {
-      //setTimeout(() => reject('Здесь какая то причина ...'), 0);
-      setTimeout(
-        () =>
-          resolve({
-            status: 'job completed',
-            result: new Date().toISOString(),
-          }),
-        100000
-      );
-    });
+    // console.log(new Date().toISOString(), 'Starting name: ' + job.name + ', job: ' + job.id);
+    // return new Promise((resolve, reject) => {
+    //   //setTimeout(() => reject('Здесь какая то причина ...'), 0);
+    //   setTimeout(
+    //     () =>
+    //       resolve({
+    //         status: 'job completed',
+    //         result: new Date().toISOString(),
+    //       }),
+    //     100000
+    //   );
+    // });
   },
   {
-    settings: {
-      stalledInterval: 300,
-    },
+    lockDuration: 900000,
     prefix,
     connection: createBullConnection('worker'), // BULL_REDIS_URI,
   }
@@ -74,5 +69,5 @@ export default {
   name: queueSettings.name as string,
   prefix: queueSettings.prefix as string,
   bullQueue: metricsQueue,
-  jobNames: ['fetch_metrics_every_5m'],
+  jobNames: ['fetch_metrics_every_5s'],
 };

example/src/index.ts

@@ -7,9 +7,13 @@ const server = new ApolloServer({
   context: () => {
     return { Queues };
   },
+  subscriptions: {
+    path: '/',
+  },
 });
 
-server.listen(5000).then(({ url }) => {
+server.listen(process.env.PORT || 5000).then(({ url }) => {
   console.log(`🚀 Server ready at ${url}`);
+  console.log(`🚀 WebSocket ready at ${url.replace(/^http/i, 'ws')}`);
   console.log(`🚀 Server pid: ${process.pid}`);
 });

src/helpers/queueEventsListen.ts

@@ -8,7 +8,7 @@ export function getAsyncIterator(
   opts: Options
 ) {
   const queueEvents = getQueueEventsSingleton(prefix, queueName, opts);
-  return createAsyncIterator(queueEvents, prefix, queueName, eventName);
+  return createAsyncIterator(queueEvents, eventName);
 }
 
 const queueEventsMap = new Map();
@@ -32,8 +32,6 @@ function getQueueEventsSingleton(prefix: string, queueName: string, opts: Option
 
 function createAsyncIterator<T = any>(
   queueEvents: QueueEvents,
-  prefix: string,
-  queueName: string,
   eventName: string
 ): Required<AsyncIterator<T>> {
   let pullSeries: any = [];
@@ -43,7 +41,7 @@ function createAsyncIterator<T = any>(
   const pushValue = async (event) => {
     if (pullSeries.length !== 0) {
       const resolver = pullSeries.shift();
-      resolver(event);
+      resolver({ value: event, done: false });
     } else {
       pushSeries.push(event);
     }
@@ -61,7 +59,7 @@ function createAsyncIterator<T = any>(
   };
 
   const handler = (event) => {
-    pushValue({ prefix, queueName, ...event });
+    pushValue(event);
   };
 
   queueEvents.on(eventName, handler);

src/subscriptions/onJobActive.ts

@@ -27,7 +27,7 @@ export function createOnJobActiveFC(
       },
       queueName: 'String!',
     },
-    resolve: async ({ prefix, queueName, jobId, prev }) => {
+    resolve: async ({ jobId, prev }, { prefix, queueName }) => {
       const queue = getQueue(prefix, queueName, opts);
       const job = await queue.getJob(jobId);
       return {

src/subscriptions/onJobCompleted.ts

@@ -26,7 +26,7 @@ export function createOnJobCompletedFC(
       },
       queueName: 'String!',
     },
-    resolve: async ({ prefix, queueName, jobId }) => {
+    resolve: async ({ jobId }, { prefix, queueName }) => {
       const queue = getQueue(prefix, queueName, opts);
       const job = await queue.getJob(jobId);
       return {

src/subscriptions/onJobDelayed.ts

@@ -27,7 +27,7 @@ export function createOnJobDelayedFC(
       },
       queueName: 'String!',
     },
-    resolve: async ({ prefix, queueName, jobId, delay }) => {
+    resolve: async ({ jobId, delay }, { prefix, queueName }) => {
       const queue = getQueue(prefix, queueName, opts);
       const job = await queue.getJob(jobId);
       return {

src/subscriptions/onJobFailed.ts

@@ -26,7 +26,7 @@ export function createOnJobFailedFC(
       },
       queueName: 'String!',
     },
-    resolve: async ({ prefix, queueName, jobId }) => {
+    resolve: async ({ jobId }, { prefix, queueName }) => {
       const queue = getQueue(prefix, queueName, opts);
       const job = await queue.getJob(jobId);
       return {

src/subscriptions/onJobProgress.ts

@@ -27,7 +27,7 @@ export function createOnJobProgressFC(
       },
       queueName: 'String!',
     },
-    resolve: async ({ prefix, queueName, jobId, data }) => {
+    resolve: async ({ jobId, data }, { prefix, queueName }) => {
       const queue = getQueue(prefix, queueName, opts);
       const job = await queue.getJob(jobId);
       return {

src/subscriptions/onJobRemoved.ts

@@ -27,7 +27,7 @@ export function createOnJobRemovedFC(
       },
       queueName: 'String!',
     },
-    resolve: async ({ prefix, queueName, jobId, prev }) => {
+    resolve: async ({ jobId, prev }, { prefix, queueName }) => {
       const queue = getQueue(prefix, queueName, opts);
       const job = await queue.getJob(jobId);
       return {

src/subscriptions/onJobStalled.ts

@@ -26,7 +26,7 @@ export function createOnJobStalledFC(
       },
       queueName: 'String!',
     },
-    resolve: async ({ prefix, queueName, jobId }) => {
+    resolve: async ({ jobId }, { prefix, queueName }) => {
       const queue = getQueue(prefix, queueName, opts);
       const job = await queue.getJob(jobId);
       return {