feat(bullmq): support ignored children (#682)

Author: roggervalf

example/bullmq_with_flows.js

@@ -57,8 +57,26 @@ async function main() {
     queueName: parentQueueName,
     data: {},
     children: [
-      {name: 'child', data: {idx: 0, foo: 'bar'}, queueName},
-      {name: 'child', data: {idx: 1, foo: 'baz'}, queueName},
+      {
+        name: 'child',
+        data: {idx: 0, foo: 'bar'},
+        queueName,
+        opts: {
+          ignoreDependencyOnFailure: true,
+          removeOnComplete: true,
+          removeOnFail: true,
+        },
+      },
+      {
+        name: 'child',
+        data: {idx: 1, foo: 'baz'},
+        queueName,
+        opts: {
+          failParentOnFailure: true,
+          removeOnComplete: true,
+          removeOnFail: true,
+        },
+      },
       {name: 'child', data: {idx: 2, foo: 'qux'}, queueName},
     ],
   });

example/package.json

@@ -20,7 +20,7 @@
     "@fastify/express": "^2.3.0",
     "bee-queue": "^1.4.0",
     "bull": "^4.12.4",
-    "bullmq": "^4.8.0",
+    "bullmq": "^5.51.0",
     "express": "^4.17.1",
     "fastify": "^4.13.0"
   }

example/yarn.lock

@@ -7,16 +7,15 @@ $(document).ready(() => {
 
   function formatToTreeView(flow, flowHost) {
     const {job, children} = flow;
+    if (!job) {
+      return {};
+    }
     const text = `${job.name} <a href="${basePath}/${encodeURIComponent(
       flowHost
     )}/${encodeURIComponent(job.queueName)}/${
       job.id
     }"><span class="label label-default">${job.id}</span></a>`;
 
-    const href = `${basePath}/${encodeURIComponent(
-      flowHost
-    )}/${encodeURIComponent(job.queueName)}/${job.id}`;
-
     if (children && children.length > 0) {
       return {
         text,

public/dashboard.js

@@ -49,21 +49,29 @@ async function handler(req, res) {
     job.parent = JobHelpers.getKeyProperties(job.parentKey);
     const processedCursor = parseInt(req.query.processedCursor, 10) || 0;
     const processedCount = parseInt(req.query.processedCount, 10) || 25;
+    const ignoredCursor = parseInt(req.query.ignoredCursor, 10) || 0;
+    const ignoredCount = parseInt(req.query.ignoredCount, 10) || 25;
     const unprocessedCursor = parseInt(req.query.unprocessedCursor, 10) || 0;
     const unprocessedCount = parseInt(req.query.unprocessedCount, 10) || 25;
     job.processedCount = processedCount;
     job.unprocessedCount = unprocessedCount;
-
+    job.ignoredCount = ignoredCount;
     const {
-      processed,
-      unprocessed,
+      processed = {},
+      ignored = {},
+      unprocessed = [],
       nextProcessedCursor,
+      nextIgnoredCursor,
       nextUnprocessedCursor,
     } = await job.getDependencies({
       processed: {
         cursor: processedCursor,
         count: processedCount,
       },
+      ignored: {
+        cursor: ignoredCursor,
+        count: ignoredCount,
+      },
       unprocessed: {
         cursor: unprocessedCursor,
         count: unprocessedCount,
@@ -73,16 +81,24 @@ async function handler(req, res) {
     job.countDependencies = count;
 
     job.processedCursor = nextProcessedCursor;
+    job.ignoredCursor = nextIgnoredCursor;
     job.unprocessedCursor = nextUnprocessedCursor;
     if (unprocessed && unprocessed.length) {
       job.unprocessedChildren = unprocessed.map((child) => {
         return JobHelpers.getKeyProperties(child);
       });
     }
 
-    if (processed) {
-      const childrenKeys = Object.keys(processed);
-      job.processedChildren = childrenKeys.map((child) => {
+    const processedKeys = Object.keys(processed);
+    if (processedKeys.length) {
+      job.processedChildren = processedKeys.map((child) => {
+        return JobHelpers.getKeyProperties(child);
+      });
+    }
+
+    const ignoredKeys = Object.keys(ignored);
+    if (ignoredKeys.length) {
+      job.ignoredChildren = ignoredKeys.map((child) => {
         return JobHelpers.getKeyProperties(child);
       });
     }

src/server/views/dashboard/jobDetails.js

@@ -10,7 +10,7 @@
         <li {{#eq currentPage 1}} class="disabled">
           <span>&laquo;</span>
           {{else}}
-          ><a
+          <a
             href="{{ basePath }}/{{ encodeURI queueHost }}/{{ encodeURI queueName }}/{{ state }}?page={{ subtract currentPage 1 }}&amp;pageSize={{ pageSize }}&amp;order={{ order }}"
             aria-label="Previous">
             <span aria-hidden="true">&laquo;</span>

src/server/views/dashboard/templates/queueJobsByState.hbs

@@ -1,12 +1,14 @@
 const Helpers = {
   processFlow: function (flow) {
+    if (!flow) return {};
     const {job, children} = flow;
+    const filteredChildren = children.filter((child) => child);
     const queueName = job.queueName;
 
-    if (children && children.length > 0) {
+    if (filteredChildren && filteredChildren.length > 0) {
       return {
         job: {...job, queueName},
-        children: children.map((child) => this.processFlow(child)),
+        children: filteredChildren.map((child) => this.processFlow(child)),
       };
     } else {
       return {

src/server/views/helpers/flowHelpers.js

@@ -227,6 +227,31 @@
   </div>
 </div>
 {{/if}}
+
+{{#if this.ignoredChildren }}
+<div class="row">
+  <div class="col-sm-12">
+    <h5>Ignored Children <span class="count">{{ this.countDependencies.ignored}}</span></h5>
+
+    <nav aria-label="Ignored navigation">
+      <ul class="pagination">
+        <li><a
+            href="{{ basePath }}/{{ encodeURI queueHost }}/{{ encodeURI queueName }}/{{ encodeURI this.id }}?ignoredCursor={{ this.ignoredCursor }}&amp;ignoredCount={{ this.ignoredCount }}"
+            aria-label="Previous">
+            <span aria-hidden="true">&raquo;</span>
+          </a>
+        </li>
+      </ul>
+    </nav>
+
+    {{#each this.ignoredChildren}}
+    <a href="{{ ../basePath }}/{{ encodeURI ../queueHost }}/{{ encodeURI this.queueName }}/{{ encodeURI this.id }}">
+      <span class="label label-warning">{{ this.id }}</span>
+    </a>
+    {{/each}}
+  </div>
+</div>
+{{/if}}
 <br>
 {{/if}}