How can i properly send a stream from forked child process in Node.js?

Question

I tried something like this in the forked child process:

WriterPlugin.js (child process)

  function sendCursor(){
                        try {
                let cursor = await getQueryCursor(reqBody, db);
                cursor.on('data', (data) => {
                    process.send(data);
                })
            } catch (error) {
                process.send({
                    message: error.toString(),
                    status: 400
                })
            }
        }

controller.js (parent process)

const childProcess = fork(fileToExec);
    childProcess.send(objectToProcess);
    childProcess.on('message', (data) => {
            reply.send(data);
    })

This one printed just last data of cursor and i faced with a fastifyError:

"code":"FST_ERR_REP_ALREADY_SENT","statusCode":500},"msg":"Reply already sent"}

How can i properly handle the cursor.stream() from a forked child process using fastify?

Answer 1

You need to push the data into a stream to accomplish this task:

const { fork } = require('child_process')
const { Readable } = require('stream')

const fastify = require('fastify')()

fastify.get('/', (request, reply) => {
  const stream = new Readable({
    read (size) {}
  })

  const childProcess = fork('./external-file.js')
  childProcess.on('message', (data) => {
    stream.push(JSON.stringify(data)) // it must be a string
  })
  childProcess.on('close', (data) => {
    stream.push(null)
  })

  reply.send(stream)
})

fastify.listen(8080)

Where external-file.js is:

let iteration = 0
const timer = setInterval(() => {
  const data = { iteration: iteration++ }
  process.send(data)

  if (iteration === 3) {
    clearInterval(timer)
  }
}, 1000)