job moves to completed when receiving data but before processing it.

[pnodet]

Hi,

I am trying to fetch a large amount of data and thought to use bull to help handling it.
I am quite new to bull but you will find below roughly how I dit it.

However, the job is moved to complete when the fetch action is done.
The job will then pipe the stream and process the data.

const worker = new Worker('downloads', async job => {
	const responseStream = await fetch(job.data.url);
	responseStream.body
		.pipe(ndjson.parse())
		.on('data', async currentGame => {
			const game = await generateGame(currentGame);
			await game.populate('user');
			await game.save()
		.on('pause', () => {
			console.log('pause');
		})
		.on('end', () => {
			console.log('end');
		})
		.on('error', error => {
			console.log(new Error(error));
		});
}, {connection});

worker.on('completed', job => {
	console.log(`${job.id} has completed!`);
});

[mull]

Your second statement, starting with responseStream.body, does not return a promise. Hence, the worker finishes immediately. You want to either return a promise or find something you can await.

[pnodet]

Hey, thanks for the answer!

Do you know where can I find documentation on this? When does the job know it should end? Is there any option to make the job open until I manually move it to completed ?

I will admit I am having a hard time implementing this.

I update the code but the completed event is still triggered before any progress.

const worker = new Worker('downloads', async job => {
    fetch(job.data.url).
        then(responseStream => {
            new Promise((resolve, reject) => {
                responseStream.body
                    .pipe(ndjson.parse())
		    .on('data', async current => {
			const game = await generateGame(currentGame);
			await game.populate('user');
			await game.save()
		    })
		    .on('end', resolve)
		    .on('error', reject);
		    })
		    .then(() => {
			console.log(`end`);
		    })
		    .catch(error => {
			console.error(error);
		    });
            });
    },
    {connection},
);

[manast]

You are still not returning the promise, try this:

const worker = new Worker(
  "downloads",
  async (job) => 
    fetch(job.data.url).then((responseStream) =>
      new Promise((resolve, reject) => {
        responseStream.body
          .pipe(ndjson.parse())
          .on("data", async (current) => {
            const game = await generateGame(currentGame);
            await game.populate("user");
            await game.save();
          })
          .on("end", resolve)
          .on("error", reject);
      })
    ),
  { connection }
);