need to close a queue?

[pcace]

Hi there, i thing i have a problem with this code. It does run and works as expected, BUT there is this one line "console.log("#######CURRENTJOBS IN ..." it will log one more time for each run of the worker. so the first time the worker finishes it logs 1 time, when it finishes the second time, this console log gets triggered 2 times. so that tells me that something is not properly closed. (or is it?) as far as i understand are the jobs removed after finished. but what is still open here? if i close the queue, workers cannot find the queue anymore. so whats wrong here?

Any help would be great!!

const connection = {
    connection: {
        host: process.env.REDIS_SERVER?.toString(),
        port: parseInt(process.env.REDIS_PORT ? process.env.REDIS_PORT.toString() : '6379'),
        password: process.env.REDIS_PASSWORD?.toString(),
    },
}
const intersectionQueue = new Queue('intersectionQueue', connection)
const queueEvents = new QueueEvents('intersectionQueue', connection)

export const runIntersectionServiceWorker = async (
    coordinatesList: { x: number; y: number; z: number }[],
    fileContent: string
): Promise<string> => {


    const intersectionJob = await intersectionQueue.add(
        'intersectionJob',
        {
            fileContent: fileContent,
            coordinatesList: coordinatesList,
        },
        { removeOnComplete: true, removeOnFail: true }
    )

    return new Promise((resolve, reject) => {
        queueEvents.on('completed', async (job) => {
            // const intersectionJob = await Job.fromId(intersectionQueue, job.jobId);
            console.log(getCurrentTime(), `DEBUG : Job Nr: ${job.jobId} name: ${intersectionJob.name}, has completed!`)
            const returnvalue = job.returnvalue
            intersectionJob.remove() //should not be necessary since the job is removeOnComplete :true !?
            console.log("#######CURRENTJOBS IN ", await intersectionQueue.getJobs()) //for each run this log comes one time more often... why!?
            // intersectionQueue.close() // cannot do that, otherways it does not get the queue on the second run.
            resolve(returnvalue)
                
        })
        queueEvents.on('failed', async job => {
               // some errorhandling
              }
        })
    })
}

[manast]

For starters, and BullMQ unrelated, you are not cleaning up the listeners that you are adding inside runIntersectionServiceWorker, so they will be listening for global events for ever.

[pcace]

Hi there, thank you! That makes toally sense!
so i just add add a await queueEvents.removeAllListeners() before adding a new one? - at least that fixes my issue i guess