feat: add paused field to queue

intpro · Boris Dorofeev · web-flow · commit 44f97e0d83e8 · 2020-05-11T18:23:35.000+06:00
* fix: change prefix discovery
* fix: bull-dot-project prefix
* feat: add paused field to queue
* fix: failed count
* refactor: according new graphql-compose syntax
* fix: redis uri

Co-authored-by: Boris Dorofeev &lt;bdorofeev@bdorofeev-laptop.corp.ps.kz&gt;
diff --git a/example/src/connectRedis.ts b/example/src/connectRedis.ts
@@ -1,7 +1,7 @@
 import Redis, { RedisOptions } from 'ioredis';
 
 // TODO: remove get from OPTIONS
-const BULL_REDIS_URI = 'redis://127.0.0.1:6379';
+const BULL_REDIS_URI = process.env.BULL_REDIS_URI || 'redis://127.0.0.1:6379';
 
 const redisInstances = new Map<string, Redis.Redis>();
 
diff --git a/example/src/demo_queues/fetchMetrics.ts b/example/src/demo_queues/fetchMetrics.ts
@@ -29,7 +29,9 @@ const metricsWorker = new Worker(
   async (job) => {
     //https://github.com/taskforcesh/bullmq/issues/69
     console.log(new Date().toISOString(), 'Starting name: ' + job.name + ', job: ' + job.id);
-    return new Promise((resolve) => {
+    return new Promise((resolve, reject) => {
+      //throw new Error('Ошибка обработки данных...');
+      //setTimeout(() => reject('Здесь какая то причина ...'), 0);
       setTimeout(
         () =>
           resolve({
diff --git a/package.json b/package.json
@@ -25,7 +25,7 @@
   },
   "homepage": "https://github.com/graphql-compose/graphql-compose-bullmq",
   "peerDependencies": {
-    "graphql-compose": "^7.0.4",
+    "graphql-compose": "^7.15.0",
     "bullmq": "^1.8.0"
   },
   "devDependencies": {
@@ -47,7 +47,7 @@
     "express": "^4.17.1",
     "express-graphql": "^0.9.0",
     "graphql": "14.6.0",
-    "graphql-compose": "7.14.0",
+    "graphql-compose": "7.15.0",
     "jest": "25.2.4",
     "prettier": "2.0.2",
     "rimraf": "3.0.2",
diff --git a/src/mutation/jobAddBulk.ts b/src/mutation/jobAddBulk.ts
@@ -13,7 +13,7 @@ export function createJobAddBulkFC(
     type: sc.createObjectTC({
       name: `${typePrefix}JobAddBulkPayload`,
       fields: {
-        jobs: getJobTC(sc, opts).getTypePlural(),
+        jobs: getJobTC(sc, opts).List,
       },
     }),
     args: {
@@ -22,30 +22,28 @@ export function createJobAddBulkFC(
         defaultValue: 'bull',
       },
       queueName: 'String!',
-      jobs: sc
-        .createInputTC({
-          name: `${typePrefix}JobAddInputBulk`,
-          fields: {
-            name: 'String!',
-            data: 'JSON!',
-            options: sc.createInputTC({
-              name: `${typePrefix}JobOptionsInputBulk`,
-              fields: {
-                priority: 'Int',
-                delay: 'Int',
-                attempts: 'Int',
-                backoff: 'JSON', // | TODO: BackoffOptions
-                lifo: 'Boolean',
-                timeout: 'Int',
-                jobId: 'String',
-                removeOnComplete: 'Boolean', //TODO: bool or int
-                removeOnFail: 'Boolean', //TODO: bool or int
-                stackTraceLimit: 'Int',
-              },
-            }),
-          },
-        })
-        .getTypePlural(),
+      jobs: sc.createInputTC({
+        name: `${typePrefix}JobAddInputBulk`,
+        fields: {
+          name: 'String!',
+          data: 'JSON!',
+          options: sc.createInputTC({
+            name: `${typePrefix}JobOptionsInputBulk`,
+            fields: {
+              priority: 'Int',
+              delay: 'Int',
+              attempts: 'Int',
+              backoff: 'JSON', // | TODO: BackoffOptions
+              lifo: 'Boolean',
+              timeout: 'Int',
+              jobId: 'String',
+              removeOnComplete: 'Boolean', //TODO: bool or int
+              removeOnFail: 'Boolean', //TODO: bool or int
+              stackTraceLimit: 'Int',
+            },
+          }),
+        },
+      }).List,
     },
     resolve: async (_, { prefix, queueName, jobs }) => {
       const queue = await findQueue(prefix, queueName, opts);
diff --git a/src/mutation/jobAddCron.ts b/src/mutation/jobAddCron.ts
@@ -37,18 +37,16 @@ export function createJobAddCronFC(
           removeOnComplete: 'Boolean', //TODO: bool or int
           removeOnFail: 'Boolean', //TODO: bool or int
           stackTraceLimit: 'Int',
-          repeat: sc
-            .createInputTC({
-              name: `${typePrefix}JobOptionsInputRepeatCron`,
-              fields: {
-                tz: 'String',
-                endDate: 'Date',
-                limit: 'Int',
-                cron: 'String!',
-                startDate: 'Date',
-              },
-            })
-            .getTypeNonNull(),
+          repeat: sc.createInputTC({
+            name: `${typePrefix}JobOptionsInputRepeatCron`,
+            fields: {
+              tz: 'String',
+              endDate: 'Date',
+              limit: 'Int',
+              cron: 'String!',
+              startDate: 'Date',
+            },
+          }).NonNull,
         },
       }),
     },
diff --git a/src/mutation/jobAddEvery.ts b/src/mutation/jobAddEvery.ts
@@ -37,17 +37,15 @@ export function createJobAddEveryFC(
           removeOnComplete: 'Boolean', //TODO: bool or int
           removeOnFail: 'Boolean', //TODO: bool or int
           stackTraceLimit: 'Int',
-          repeat: sc
-            .createInputTC({
-              name: `${typePrefix}JobOptionsInputRepeatEvery`,
-              fields: {
-                tz: 'String',
-                endDate: 'Date',
-                limit: 'Int',
-                every: 'String!',
-              },
-            })
-            .getTypeNonNull(),
+          repeat: sc.createInputTC({
+            name: `${typePrefix}JobOptionsInputRepeatEvery`,
+            fields: {
+              tz: 'String',
+              endDate: 'Date',
+              limit: 'Int',
+              every: 'String!',
+            },
+          }).NonNull,
         },
       }),
     },
diff --git a/src/query/queueKeys.ts b/src/query/queueKeys.ts
@@ -8,15 +8,13 @@ export function createQueueKeysFC(
 ): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
   const { typePrefix } = opts;
   return {
-    type: sc
-      .createObjectTC({
-        name: `${typePrefix}QueueKeysResult`,
-        fields: {
-          queueName: 'String!',
-          prefix: 'String!',
-        },
-      })
-      .getTypePlural(),
+    type: sc.createObjectTC({
+      name: `${typePrefix}QueueKeysResult`,
+      fields: {
+        queueName: 'String!',
+        prefix: 'String!',
+      },
+    }).List,
     args: {
       prefixGlob: {
         type: 'String',
diff --git a/src/query/queues.ts b/src/query/queues.ts
@@ -8,7 +8,7 @@ export function createQueuesFC(
   opts: Options
 ): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
   return {
-    type: getQueueTC(sc, opts).getTypeNonNull().getTypePlural(),
+    type: getQueueTC(sc, opts).NonNull.List,
     args: {
       prefix: {
         type: 'String',
diff --git a/src/types/job/Job.opts.repeat.ts b/src/types/job/Job.opts.repeat.ts
@@ -44,8 +44,5 @@ export function createRepeatOptionsTC(sc: SchemaComposer<any>, opts: Options) {
     return isObject(value) && value.hasOwnProperty('cron');
   });
 
-  sc.addSchemaMustHaveType(RepeatOptionsEveryTC);
-  sc.addSchemaMustHaveType(RepeatOptionsCronTC);
-
   return RepeatOptionsInterfaceTC;
 }
diff --git a/src/types/job/JobInput.ts b/src/types/job/JobInput.ts
@@ -17,7 +17,7 @@ export function createJobDataITC(
       someTC = someTC.getInputTypeComposer();
     }
 
-    return someTC.getTypeNonNull();
+    return someTC.NonNull;
   } else {
     return 'JSON!';
   }
diff --git a/src/types/queue/Queue.activeJobs.ts b/src/types/queue/Queue.activeJobs.ts
@@ -8,7 +8,7 @@ export function createActiveJobsFC(
   opts: Options
 ): ObjectTypeComposerFieldConfigDefinition<any, any> {
   return {
-    type: getJobTC(sc, opts).getTypePlural(),
+    type: getJobTC(sc, opts).List,
     args: {
       start: {
         type: 'Int',
diff --git a/src/types/queue/Queue.completedJobs.ts b/src/types/queue/Queue.completedJobs.ts
@@ -8,7 +8,7 @@ export function createCompletedJobsFC(
   opts: Options
 ): ObjectTypeComposerFieldConfigDefinition<any, any> {
   return {
-    type: getJobTC(sc, opts).getTypePlural(),
+    type: getJobTC(sc, opts).List,
     args: {
       start: {
         type: 'Int',
diff --git a/src/types/queue/Queue.delayedJobs.ts b/src/types/queue/Queue.delayedJobs.ts
@@ -8,7 +8,7 @@ export function createDelayedJobsFC(
   opts: Options
 ): ObjectTypeComposerFieldConfigDefinition<any, any> {
   return {
-    type: getJobTC(sc, opts).getTypePlural(),
+    type: getJobTC(sc, opts).List,
     args: {
       start: {
         type: 'Int',
diff --git a/src/types/queue/Queue.failedJobs.ts b/src/types/queue/Queue.failedJobs.ts
@@ -8,7 +8,7 @@ export function createFailedJobsFC(
   opts: Options
 ): ObjectTypeComposerFieldConfigDefinition<any, any> {
   return {
-    type: getJobTC(sc, opts).getTypePlural(),
+    type: getJobTC(sc, opts).List,
     args: {
       start: {
         type: 'Int',
diff --git a/src/types/queue/Queue.isPaused.ts b/src/types/queue/Queue.isPaused.ts
@@ -0,0 +1,13 @@
+import { Queue } from 'bullmq';
+import { ObjectTypeComposerFieldConfigDefinition } from 'graphql-compose';
+
+export function createIsPausedFC(): ObjectTypeComposerFieldConfigDefinition<any, any> {
+  return {
+    type: 'Boolean',
+    resolve: async (queue: Queue) => {
+      const client = await queue.client;
+      const meta = await client.hgetall(queue.keys.meta);
+      return meta?.paused ? !!+meta.paused : false;
+    },
+  };
+}
diff --git a/src/types/queue/Queue.jobCounts.ts b/src/types/queue/Queue.jobCounts.ts
@@ -25,7 +25,7 @@ export function createJobCountFC(
         JobStatusEnum.ACTIVE,
         JobStatusEnum.COMPLETED,
         JobStatusEnum.DELAYED,
-        JobStatusEnum.PAUSED,
+        JobStatusEnum.FAILED,
         JobStatusEnum.WAITING
       );
     },
diff --git a/src/types/queue/Queue.jobs.ts b/src/types/queue/Queue.jobs.ts
@@ -9,7 +9,7 @@ export function createJobsFC(
   opts: Options
 ): ObjectTypeComposerFieldConfigDefinition<any, any> {
   return {
-    type: getJobTC(sc, opts).getTypePlural(),
+    type: getJobTC(sc, opts).List,
     args: {
       status: getJobStatusEnumTC(sc, opts),
       start: {
diff --git a/src/types/queue/Queue.repeatables.ts b/src/types/queue/Queue.repeatables.ts
@@ -9,21 +9,19 @@ export function createRepeatablesFC(
   const { typePrefix } = opts;
 
   return {
-    type: sc
-      .createObjectTC({
-        name: `${typePrefix}RepeatableJobInformation`,
-        fields: {
-          key: 'String',
-          name: 'String',
-          id: 'String',
-          endDate: 'Date',
-          tz: 'String',
-          cron: 'String',
-          //every: 'Date', //TODO: вроде как должен быть обязательным, проверить - нет в бул-4
-          next: 'Date',
-        },
-      })
-      .getTypePlural(),
+    type: sc.createObjectTC({
+      name: `${typePrefix}RepeatableJobInformation`,
+      fields: {
+        key: 'String',
+        name: 'String',
+        id: 'String',
+        endDate: 'Date',
+        tz: 'String',
+        cron: 'String',
+        //every: 'Date', //TODO: вроде как должен быть обязательным, проверить - нет в бул-4
+        next: 'Date',
+      },
+    }).List,
     resolve: async (queue: Queue) => {
       return queue.getRepeatableJobs();
     },
diff --git a/src/types/queue/Queue.ts b/src/types/queue/Queue.ts
@@ -1,3 +1,4 @@
+import { createIsPausedFC } from './Queue.isPaused';
 import { createJobCountFC } from './Queue.jobCounts';
 import { createRepeatablesFC } from './Queue.repeatables';
 import { createJobsFC } from './Queue.jobs';
@@ -16,6 +17,7 @@ export function getQueueTC(sc: SchemaComposer<any>, opts: Options) {
   return sc.getOrCreateOTC(`${typePrefix}Queue`, (etc) => {
     etc.addFields({
       name: 'String!',
+      isPaused: createIsPausedFC(),
       jobCounts: createJobCountFC(sc, opts),
       repeatables: createRepeatablesFC(sc, opts),
       jobs: createJobsFC(sc, opts),
diff --git a/src/types/queue/Queue.waitingJobs.ts b/src/types/queue/Queue.waitingJobs.ts
@@ -8,7 +8,7 @@ export function createWaitingJobsFC(
   opts: Options
 ): ObjectTypeComposerFieldConfigDefinition<any, any> {
   return {
-    type: getJobTC(sc, opts).getTypePlural(),
+    type: getJobTC(sc, opts).List,
     args: {
       start: {
         type: 'Int',
diff --git a/src/types/queue/Queue.workers.ts b/src/types/queue/Queue.workers.ts
@@ -9,32 +9,30 @@ export function createWorkersTC(
   const { typePrefix } = opts;
 
   return {
-    type: sc
-      .createObjectTC({
-        // see https://redis.io/commands/client-list
-        name: `${typePrefix}QueueWorker`,
-        fields: {
-          id: 'String',
-          addr: 'String',
-          fd: 'String',
-          name: 'String',
-          age: 'Int',
-          idle: 'Int',
-          flags: 'String',
-          db: 'Int',
-          sub: 'Int',
-          psub: 'Int',
-          multi: 'Int',
-          qbuf: 'Int',
-          qbufFree: 'Int',
-          obl: 'Int',
-          oll: 'Int',
-          omem: 'Int',
-          events: 'String',
-          cmd: 'String',
-        },
-      })
-      .getTypePlural(),
+    type: sc.createObjectTC({
+      // see https://redis.io/commands/client-list
+      name: `${typePrefix}QueueWorker`,
+      fields: {
+        id: 'String',
+        addr: 'String',
+        fd: 'String',
+        name: 'String',
+        age: 'Int',
+        idle: 'Int',
+        flags: 'String',
+        db: 'Int',
+        sub: 'Int',
+        psub: 'Int',
+        multi: 'Int',
+        qbuf: 'Int',
+        qbufFree: 'Int',
+        obl: 'Int',
+        oll: 'Int',
+        omem: 'Int',
+        events: 'String',
+        cmd: 'String',
+      },
+    }).List,
     args: {},
     resolve: async (queue: Queue) => {
       const workers = await queue.getWorkers();
diff --git a/yarn.lock b/yarn.lock

Original file line number	Diff line number	Diff line change
`@@ -44,8 +44,5 @@ export function createRepeatOptionsTC(sc: SchemaComposer<any>, opts: Options) {`
`44`	`44`	`return isObject(value) && value.hasOwnProperty('cron');`
`45`	`45`	`});`
`46`	`46`
`47`		`- sc.addSchemaMustHaveType(RepeatOptionsEveryTC);`
`48`		`- sc.addSchemaMustHaveType(RepeatOptionsCronTC);`
`49`		`-`
`50`	`47`	`return RepeatOptionsInterfaceTC;`
`51`	`48`	`}`
Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@ export function createJobDataITC(`
`17`	`17`	`someTC = someTC.getInputTypeComposer();`
`18`	`18`	`}`
`19`	`19`
`20`		`- return someTC.getTypeNonNull();`
	`20`	`+ return someTC.NonNull;`
`21`	`21`	`} else {`
`22`	`22`	`return 'JSON!';`
`23`	`23`	`}`