Access task in queue by id

[diamondio]

Access task in queue by id

[leanderlee]

You can use better-queue to trigger those tasks, but perhaps you can keep the tasks in your own dictionary, for example:

const Queue = require('better-queue');
const tasks = {};
tasks['task-1'] = { ... };
const q = new Queue(function (id, cb) {
  const task = tasks[id];
  // Trigger some server procedure
  cb();
});
q.push('task-1');

// Later in your server
server.handler('/', function () {
  // Edit the task to mark it as complete
})

[leanderlee]

One way you can do this is keep the cb in memory, for example:

const Queue = require('better-queue');
const cbs = {};
const q = new Queue(function (task, cb) {
  const { id } = task;
  cbs[id] = cb;
  // Trigger some server procedure, eg:
  socket.emit({ id, ... });
});
q.push({ id: 'my-id', ... });

socket.on('receive-mail', (data) => {
  const { id, result } = data;
  const cb = cbs[id];
  cb(null, result);
  cbs[id] = null; // Remove the cb to prevent memory leaks!
});

For distributed systems you would probably need to keep the status of your tasks via a database.

[leanderlee]

There's currently no way to trigger these outside of the processing function. What would be the use case for this?

[karmac2015]

I want to transfer files between two servers , I use socket.io stream for this purpose. When user needs to submit file I add the task to the queue, then after receiving custom event about (progress, finish ...) I want to trigger task's methods in queue.

Now I use the following code to access tasks:
myQueue._workers[taskId].progressBatch(uploadedBytes, fileSize,"uploading")
and
myQueue._workers[taskId].finishBatch("File was sent successfully")

[leanderlee]

It's not recommended to use _ functions, as they might change when the implementation changes.

But if you must get it to work this way, you can access the taskIds through this._taskIds in the processing function. I would recommend you create your own ID/progress system as it sounds like your needs are quite specific to your socket implementation.

[karmac2015]

Suppose I want to create my own array that contains all my tasks, what is the best way to always check if there are tasks in my array that need processing ? what is the technology used in your library?
I like your library but I think you need to add a way to access tasks from outside processing function.

[karmac2015]

Suppose

socket.on('receive-mail', (data) => {
      uploader._workers[data.taskId].finishBatch(`Was Sent successfully`);
});

How can I convert this part of code to edit the task to mark it as complete.

I know it is another question, but I want to know how can I use cb() inside processing function?

Thank you for help :)

[karmac2015]

Well, I understand your concept but how can I return an error using callback cb(null, result) to make processing function trigger a failed event?
And how can I solve the problem of progress event:

socket.on('progress-mail', (data) => {
      uploader._workers[data.taskId].progressBatch(data.uploadedBytes, data.fileSize, "uploading");
});

[karmac2015]

const Queue = require('better-queue');
const tasks= {};
const q = new Queue(function (task, cb) {
  const { id } = task;
  tasks[id] = task;
  // Trigger some server procedure, eg:
  socket.emit({ id, ... });
});
q.push({ id: 'my-id', ... });

socket.on('receive-mail', (data) => {
  const { id, result } = data;
  const task= tasks[id];
  // and now use task method here
 // task.progressTask() or task.finishTask() or task.failedTask()
});

What do you think about that ? Can I use this concept?
Thank you again.

[leanderlee]

You could keep track of the context (this) in the processing function... In the code you provided, instead of saving task you could save this.

The convention in NodeJS is that the first argument of a callback, such as cb(null, result) is for an error, so if you wanted to trigger a failed event you would call cb(error).

[karmac2015]

Thank you, yes it was a mistake in my code I should put (this) instead of task. I think it is better to put the task's methods in the documentation because (progressTask, finishTask and failedTask) is not listed in the documentation, I saw it in your library.

[leanderlee]

It is here in advanced.