feat(queue): support updating queue configuration (#687)

Author: roggervalf

example/bullmq.js

@@ -37,7 +37,7 @@ async function main() {
       }
     },
     {
-      concurrency: 3,
+      concurrency: 10,
       connection: {port: REDIS_SERVER_PORT},
     }
   );

public/dashboard.js

@@ -286,6 +286,48 @@ $(document).ready(() => {
     }
   });
 
+  $('.js-toggle-update-queue-meta').on('click', function () {
+    const updateMetaText = $('.js-toggle-update-queue-meta').text();
+    const shouldNotHide = updateMetaText === 'Update';
+    const newUpdateMetaText = shouldNotHide ? 'Cancel' : 'Update';
+    $('.meta-config-editor').toggleClass('hide', !shouldNotHide);
+    $('.js-toggle-update-queue-meta').text(newUpdateMetaText);
+  });
+
+  $('.js-update-queue-meta').on('click', function () {
+    const {queueHost, queueName} = window.arenaInitialPayload;
+    const concurrency = $('input.js-update-meta-concurrency').val() || null;
+    const max = $('input.js-update-meta-rl-max').val() || null;
+    const duration = $('input.js-update-meta-rl-duration').val() || null;
+
+    const stringifiedData = JSON.stringify({
+      concurrency: concurrency ? parseInt(concurrency, 10) : null,
+      max: max ? parseInt(max, 10) : null,
+      duration: duration ? parseInt(duration, 10) : null,
+    });
+
+    const response = window.confirm(
+      `Are you sure you want to update the queue "${queueHost}/${queueName}" configuration?`
+    );
+    if (response) {
+      $.ajax({
+        url: `${basePath}/api/queue/${encodeURIComponent(
+          queueHost
+        )}/${encodeURIComponent(queueName)}/update-meta`,
+        type: 'PUT',
+        data: stringifiedData,
+        contentType: 'application/json',
+      })
+        .done(() => {
+          window.location.reload();
+        })
+        .fail((jqXHR) => {
+          window.alert(`Request failed, check console for error.`);
+          console.error(jqXHR.responseText);
+        });
+    }
+  });
+
   $('.js-toggle-add-flow-editor').on('click', function () {
     const addFlowText = $('.js-toggle-add-flow-editor').text();
     const shouldNotHide = addFlowText === 'Add Flow';

src/server/views/api/bulkAction.js

@@ -40,7 +40,12 @@ function bulkAction(action) {
         await Promise.all(actionPromises);
         return res.sendStatus(200);
       } else if (action === 'clean') {
-        await queue.clean(1000, queueState);
+        if (queue.IS_BULLMQ) {
+          await queue.clean(0, 1000, queueState);
+        } else {
+          await queue.clean(1000, queueState);
+        }
+
         return res.sendStatus(200);
       }
     } catch (e) {

src/server/views/api/index.js

@@ -14,6 +14,7 @@ const bulkJobsRemove = require('./bulkJobsRemove');
 const bulkJobsRetry = require('./bulkJobsRetry');
 const queuePause = require('./queuePause');
 const queueResume = require('./queueResume');
+const queueUpdateMeta = require('./queueUpdateMeta');
 
 router.post('/queue/:queueHost/:queueName/job', jobAdd);
 router.post('/flow/:flowHost/:connectionName/flow', addFlow);
@@ -30,6 +31,7 @@ router.put('/queue/:queueHost/:queueName/job/:id/data', jobDataUpdate);
 router.patch('/queue/:queueHost/:queueName/job/:id', jobRetry);
 router.put('/queue/:queueHost/:queueName/pause', queuePause);
 router.put('/queue/:queueHost/:queueName/resume', queueResume);
+router.put('/queue/:queueHost/:queueName/update-meta', queueUpdateMeta);
 router.delete('/queue/:queueHost/:queueName/job/:id', jobRemove);
 router.delete('/queue/:queueHost/:queueName/jobs/bulk', bulkJobsClean);
 

src/server/views/api/queueUpdateMeta.js

@@ -0,0 +1,37 @@
+async function handler(req, res) {
+  const {queueName, queueHost} = req.params;
+  const data = req.body;
+
+  const {Queues} = req.app.locals;
+
+  const queue = await Queues.get(queueName, queueHost);
+
+  if (!queue) return res.status(404).json({error: 'queue not found'});
+
+  try {
+    if (queue.setGlobalConcurrency && queue.setGlobalRateLimit) {
+      if (data.concurrency !== null && data.concurrency !== undefined) {
+        await queue.setGlobalConcurrency(data.concurrency);
+      } else {
+        await queue.removeGlobalConcurrency();
+      }
+
+      if (
+        data.max !== null &&
+        data.max !== undefined &&
+        data.duration !== null &&
+        data.duration !== undefined
+      ) {
+        await queue.setGlobalRateLimit(data.max, data.duration);
+      } else {
+        await queue.removeGlobalRateLimit();
+      }
+    }
+  } catch (err) {
+    console.log('err', err);
+    return res.status(500).json({error: err.message});
+  }
+  return res.sendStatus(200);
+}
+
+module.exports = handler;

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

@@ -89,10 +89,40 @@
     {{#if globalConfig}}
     <div class="panel panel-default">
       <div class="panel-heading">
+        <div class="js-toggle-update-queue-meta btn btn-default btn-xs pull-right">Update</div>
         <h4 class="panel-title">Global Configuration</h4>
       </div>
       <div class="panel-body">
-        <table class="table">
+        <div class="meta-config-editor panel-body hide overflow-hidden">
+          <div class="form-group row">
+            <label for="queue-concurrency-input-search" class="col-sm-8 col-form-label">Concurrency</label>
+            <div class="col-sm-4">
+              <input type="number" class="js-update-meta-concurrency form-control input-sm" style="margin-right: 10px;"
+                placeholder="Concurrency"
+              >
+            </div>
+          </div>
+          <div class="form-group row">
+            <label for="queue-rl-max-input-search" class="col-sm-8 col-form-label">Rate Limit Max</label>
+            <div class="col-sm-4">
+              <input type="number" class="js-update-meta-rl-max form-control input-sm" style="margin-right: 10px;"
+                placeholder="Max"
+              >
+            </div>
+          </div>
+          <div class="form-group row">
+            <label for="queue-rl-duration-input-search" class="col-sm-8 col-form-label">Rate Limit Duration</label>
+            <div class="col-sm-4">
+              <input type="number" class="js-update-meta-rl-duration form-control input-sm" style="margin-right: 10px;"
+                placeholder="Duration"
+              >
+            </div>
+          </div>
+          <div class="js-update-queue-meta btn btn-primary btn-sm">
+            Update
+          </div>
+        </div>
+        <table class="meta-table table">
           {{#each globalConfig}}
           <tr>
             <th>{{ @key }}</th>