Warm tip: This article is reproduced from serverfault.com, please click

Node JS: Passing Response Object to Bull Queue for Server Side Events

发布于 2020-11-28 07:56:28

I am stuck at an architectural decision. I have Node + Express app, It has an API to upload files. After the upload is done, the response is closed and uploaded file is processed by FFMPEG batch wise with the help of Bull Queue + Redis. This structure works fine but recently I started testing on Server side events to give updates about processing to the end user. But I am unable to pass the response object to Bull Queue to write regular updates from server to user.

1. Imports

import childProcess from 'child_process';
import Bull from 'bull'
const Queue = new Bull('background_job', {redis: {port: process.env.port, host: process.env.host, password: process.env.password}});

2. Upload function

const uploadVideo = async(req, res) => {
    try{
        const result = await authUser(req);
        const result2 = await checkUploadFile(result);
        const result3 = await insertPost(result2, res);
        await Queue.add(result3.data, result3.opts)
    } catch(err){
        res.status(403).send(err);
    }
}

3. Promises

const authUser = (req) => {
    return new Promise((resolve, reject) => {
      //do some work
    })
}

const checkUploadFile = (result) => {
    return new Promise((resolve, reject) => {
      //do some more work
    })
}

const insertPost= (result, res) => {
    return new Promise((resolve, reject) => {
      //do final work
       ...........
      //preparing server side events
       const headers = {
            'Content-Type': 'text/event-stream',
            'Connection': 'keep-alive',
            'Cache-Control': 'no-cache',
            'Access-Control-Allow-Origin': '*'
        };
        res.writeHead(200, headers);
        res.write(JSON.stringify({status: true, id: 1})); //testing server side events for the first time

        //Let's continue to Bull
        const data = {res: res} <- error here: TypeError: Converting circular structure to JSON 
        const opts = {removeOnComplete: true, removeOnFail: true}
        resolve({data: data, opts: opts});
    })
}

4. Queue Process

Queue.process((job, done) => {
    const res = job.data.res
    childProcess.execFile('someScript.sh', [`some`, `arguments`], { stdio: ['pipe', 'pipe', 'ignore']}, (err, stderr, stdout) => {
        if(err){
            done(new Error("Failed: " + err))
            res.write(JSON.stringify({status: true, id: 2})); //here using SSE
            res.end()
        } else {
            done()
            res.write(JSON.stringify({status: false})); //here using SSE
            res.end()
        }
    })
})

5. Error logged by PM2

TypeError: Converting circular structure to JSON
    --> starting at object with constructor 'Socket'
    |     property 'parser' -> object with constructor 'HTTPParser'
    --- property 'socket' closes the circle

I tried to use JSON.stringify(res) to pass the response object as JSON but that didn't work either. Now I'm considering if this approach is right or should I go with Socket.io (which is an overkill for a simple server side events)

Thank you

Questioner
Abhinandan Chakraborty
Viewed
0
kg99 2020-11-28 19:31:28

Why do you even write this line:

const data = {res: res} <- error here: TypeError: Converting circular structure to JSON.

You still have access to the response object in uploadVideo function where you call insertPost. so it can simply be:

await Queue.add(res, result3.opts).

For instance:

const uploadVideo = async(req, res) => {
    try{
        const result = await authUser(req);
        const result2 = await checkUploadFile(result);
        const result3 = await insertPost(result2, res);
        await Queue.add(res, result3.opts); // still have access to res
    } catch(err){
        res.status(403).send(err);
    }

Remove this line:

const data = {res: res} <- error here: TypeError: Converting circular structure to JSON 

Just use response

Queue.process((res, done) => {
    //const res = job.data.res
    childProcess.execFile('someScript.sh', [`some`, `arguments`], { stdio: ['pipe', 'pipe', 'ignore']}, (err, stderr, stdout) => {
        if(err){
            done(new Error("Failed: " + err))
            res.write(JSON.stringify({status: true, id: 2})); //here using SSE
            res.end()
        } else {
            done()
            res.write(JSON.stringify({status: false})); //here using SSE
            res.end()
        }
    })
});

Edit:

I see what you mean. Had a look at the bull module. Why cant you do something like this.

const uploadVideo = async(req, res) => {
  try{
      res.jobId = 0; // we need a way to know if job completed is our request          const result = await authUser(req);
      const result2 = await checkUploadFile(result);
      const result3 = await insertPost(result2, res);
      Queue.add({id: res.jobId, somedatafromresult3: 'result3.somedata' }, result3.opts);
      Queue.on("completed", (err, data) => {
        if (data.id === res.jobId) { // check to see if completed job is our one.
          res.write(JSON.stringify(data)); //here using SSE
          res.end()
        }
        console.log(data);
      });
  } catch(err){
      res.status(403).send(err);
  }
}

Then in your process function, simply return the data which will be emitted. i.e

  videoQueue.process(function(job, done){
  childProcess.execFile('someScript.sh', [`some`, `arguments`], { stdio: ['pipe', 'pipe', 'ignore']}, (err, stderr, stdout) => {
    if(err){
        done(err, {status: true, id: job.data.id});
    } else {
      done(null, {status: false, id: job.data.id});
    }
})
})

;