How to pass data from job to another?

[apoorv569]

I was using node-cron, but it felt limited so looking around I found couple of things and I couldn't decide, but for now I'm trying BullMQ out.

I could not find much real life example for using Bull or BullMQ and the documentation is not the best either.

If a queue has multiple jobs, say job1, job2, job3 etc, and say job1 runs and returns some data that the job2 needs to run and then returns some data that job3 needs to run. How do you pass this data from job to job?

From what I understand, you would have something like this,

const conn = {
    host: "localhost",
    port: 6379
};

const process_queue = new Queue("process", { connection: conn });

async function add_jobs(queue: Queue) {
    await queue.add("get_media_files", 1);
    await queue.add("extract_metadata", 2);
    await queue.add("save_metadata", 3);

    console.log('Jobs have been added to the queue');
}

const worker = new Worker("process", async (job) => {
    if (job.name === "get_media_files") {
        logger.info(`Starting ${job.name} job`);
    }

    if (job.name === "extract_metadata") {
        logger.info(`Starting ${job.name} job`);
    }

    if (job.name === "save_metadata") {
        logger.info(`Starting ${job.name} job`);
    }

    logger.info(`Processing job: ${job.name}`);
    logger.info(`Job data: ${job.data}`);

    logger.info(`Job ${job.name} with ID ${job.id} is done`);
}, { connection: conn });

function start_jobs() {
    add_jobs(process_queue).catch((error) => {
        logger.error(`Error adding jobs: ${error}`);
    }).finally(() => {
        logger.info('Job setup complete');
    });
}

here get_media_files will return list of media files as string[] which then the extract_metadata job needs to go over and finally return the list of metadata, which then the save_metadata job needs which saves all the metadata to a DB.

how would you pass that data? also I'm not sure what initial data you pass to each job?

I basically want to run tasks on schedule and get the state, status etc from the tasks.

[manast]

Some documentation: https://docs.bullmq.io/guide/flows
Tutorials: https://blog.taskforce.sh/splitting-heavy-jobs-using-bullmq-flows/

[apoorv569]

This seems to work,

const process_queue = new Queue("process", { connection });

const worker = new Worker("process", async (job) => {
    if (job.name === "scan_directory") {
        logger.info(`Starting ${job.name} job`);

        const files = get_audio_files();

        await process_queue.add("extract_metadata", { files });
    }

    if (job.name === "extract_metadata") {
        logger.info(`Starting ${job.name} job`);

        const files = job.data.files;

        const metadata_list: Metadata[] = [];

        if (files) {
            for (const file of files) {
                const metadata = await extract_metadata(file);

                if (metadata) {
                    metadata_list.push(metadata);
                }
            }
        }

        await process_queue.add("save_metadata", { metadata_list });
    }

    if (job.name === "save_metadata") {
        logger.info(`Starting ${job.name} job`);

        const metadata_list = job.data.metadata_list;

        await save_metadata(metadata_list);
    }
}, { connection });

worker.on('active', (job) => {
    logger.info(`Job "${job.name}" with id "${job.id}" is being processed`);
});

worker.on('completed', (job) => {
    logger.info(`Job "${job.name}" with id "${job.id}" completed`);
});

worker.on('failed', (job, error) => {
    logger.error(`Job "${job?.name}" with id "${job?.id}" failed: ${error.message}`);
});

async function add_jobs() {
    await process_queue.add("scan_directory", {});
}

export function start_processing_media_files() {
    add_jobs().catch((error) => {
        logger.error(`Error adding jobs: ${error}`);
    }).finally(() => {
        logger.info('Job setup complete');
    });
}

But is this the correct way of handling this?

If I were to use FlowProducer I think my code would look something like,

const queue_opts = {
    connection,
    prefix: "sonora_",
    defaultJobOptions: {
        attempts: 3,
        backoff: {
            type: 'exponential',
            delay: 1000
        }
    }
};

const directory_scan_queue = new Queue("directory_scan", queue_opts);
const metadata_extractor_queue = new Queue("metadata_extractor", queue_opts);
const save_metadata_to_db_queue = new Queue("save_metadata_to_db", queue_opts);

const directory_scan_worker = new Worker("directory_scan", async (job) => {
    const files = get_audio_files();

    return { files }
}, { connection });

const metadata_extractor_worker = new Worker("metadata_extractor", async (job) => {
    const child_job_value = await job.getChildrenValues();

    const files = Object.values(child_job_value)[0].files;

    const metadata_list: Metadata[] = [];

    if (files) {
        for (const file of files) {
            const metadata = await extract_metadata(file);

            if (metadata) {
                metadata_list.push(metadata);
            }
        }
    }

    return { metadata_list }
}, { connection });

const save_metadata_to_db_worker = new Worker("save_metadata_to_db", async (job) => {
    const child_job_value = await job.getChildrenValues();

    const metadata_list = Object.values(child_job_value)[0].metadata_list;

    if (metadata_list) {
        await save_metadata(metadata_list);
    }
}, { connection });

const workers = [
    directory_scan_worker,
    metadata_extractor_worker,
    save_metadata_to_db_worker
];

for (const worker of workers) {
    worker.on('active', (job) => {
        logger.info(`Job ${job.name} with id ${job.id} is being processed with ${worker.name} worker`);
    });

    worker.on('completed', (job) => {
        logger.info(`Job ${job.name} with id ${job.id} completed via ${worker.name} worker`);
    });

    worker.on('failed', (job, error) => {
        logger.error(`Job ${job?.name} with id ${job?.id} failed from ${worker.name} worker: ${error.message}`);
    });
}

const flow = new FlowProducer();

async function add_jobs() {
    process.on("SIGTERM", async () => {
        await directory_scan_queue.close();
        await metadata_extractor_queue.close();
        await save_metadata_to_db_queue.close();

        logger.info("All queues closed");
    });

    flow.add({
        name: "save_metadata",
        queueName: "save_metadata_to_db",
        children: [
            {
                name: "extract_metadata",
                queueName: "metadata_extractor",
                children: [
                    {
                        name: "scan_directory",
                        queueName: "directory_scan",
                    }
                ]
            },
        ]
    });
}

export function start_processing_media_files() {
    add_jobs().catch((error) => {
        logger.error(`Error adding jobs: ${error}`);
    }).finally(() => {
        logger.info('Job setup complete');
    });
}

Am I on the right track? Do I really need different queues and workers here? Sorry I'm new to queue jobs thing.