Comment puis-je m'assurer qu'un travail ne s'exécute pas deux fois dans Bull?

J'ai deux fonctions, scheduleScan()et scan().

scan()appelle scheduleScan() lorsqu'il n'y a rien d'autre à faire à part planifier un nouveau scan , donc scheduleScan()peut planifier un scan(). Mais il y a un problème, certains travaux s'exécutent deux fois.

Je veux m'assurer qu'un seul travail est en cours de traitement à un moment donné. Comment puis-je y parvenir? Je crois que cela a quelque chose à voir avec done()(il était dans scan (), supprimé maintenant) mais je n'ai pas pu trouver de solution.

Version Bull: 3.12.1

Modification tardive importante: scan() appelle d'autres fonctions et elles peuvent ou non appeler d'autres fonctions, mais ce sont toutes des fonctions de synchronisation, donc elles n'appellent une fonction que lorsque leurs propres travaux sont terminés, il n'y a qu'une seule façon d'avancer. À la fin de "l'arborescence", je l'appelle, la dernière fonction appelle ScheduleScan (), mais il ne peut pas y avoir deux tâches simultanées en cours d'exécution. Soit dit scan()en passant, chaque travail unique se termine parscheduleScan(stock, period, milliseconds, 'called by file.js')

export function update(job) {
  // does some calculations, then it may call scheduleScan() or
  // it may call another function, and that could be the one calling
  // scheduleScan() function.
  // For instance, a function like finalize()
}

export function scan(job) {
  update(job)
}


import moment from 'moment'
import stringHash from 'string-hash'
const opts = { redis: { port: 6379, host: '127.0.0.1', password: mypassword' } }
let queue = new Queue('scan', opts)

queue.process(1, (job) => {
  job.progress(100).then(() => {
    scan(job)
  })
})

export function scheduleScan (stock, period, milliseconds, triggeredBy) {
  let uniqueId = stringHash(stock + ':' + period)

  queue.getJob(uniqueId).then(job => {
    if (!job) {
      if (milliseconds) {
        queue.add({ stock, period, triggeredBy }, { delay: milliseconds, jobId: uniqueId }).then(() => {
          // console.log('Added with ms: ' + stock + ' ' + period)
        }).catch(err => {
          if (err) {
            console.log('Can not add because it exists ' + new Date())
          }
        })
      } else {
        queue.add({ stock, period, triggeredBy }, { jobId: uniqueId }).then(() => {
          // console.log('Added without ms: ' + stock + ' ' + period)
        }).catch(err => {
          if (err) {
            console.log('Can not add because it exists ' + new Date())
          }
        })
      }
    } else {
      job.getState().then(state => {
        if (state === 'completed') {
          job.remove().then(() => {
            if (milliseconds) {
              queue.add({ stock, period, triggeredBy }, { delay: milliseconds, jobId: uniqueId }).then(() => {
                // console.log('Added with ms: ' + stock + ' ' + period)
              }).catch(err => {
                if (err) {
                  console.log('Can not add because it exists ' + new Date())
                }
              })
            } else {
              queue.add({ stock, period, triggeredBy }, { jobId: uniqueId }).then(() => {
                // console.log('Added without ms: ' + stock + ' ' + period)
              }).catch(err => {
                if (err) {
                  console.log('Can not add because it exists ' + new Date())
                }
              })
            }
          }).catch(err => {
            if (err) {
              // console.log(err)
            }
          })
        }
      }).catch(err => {
        // console.log(err)
      })
    }
  })
}

Le problème, je crois, c'est que votre scanfonction est asynchrone. Votre job.progressfonction appelle donc scan, puis appelle immédiatement, donepermettant à la file d'attente de traiter un autre travail.

Une solution pourrait être de passer le donerappel comme paramètre à vos scanet scheduleScanfonctions, et l' appeler, une fois que vous avez terminé votre travail (ou en cas d' erreur).

Une autre (meilleure) solution pourrait être de vous assurer de toujours renvoyer un Promisede scanet d' scheduleScanattendre la promesse de résolution, puis d'appeler done. Si vous faites cela, assurez-vous d'enchaîner tous vos retours de promesse dans votre scheduleScanfonction.

queue.process(1, (job, done) => {
  job.progress(100).then(() => {
    scan(job)
        .then(done)
        .catch(done)
  })
})

export function scan() {
   // business logic
   return scheduleScan()
}

// Chain all of your promise returns. Otherwise
// the scan function will return sooner and allow done to be called
// prior to the scheduleScan function finishing it's execution
export function scheduleScan() {
    return queue.getJob(..).then(() => {
        ....
        return queue.add()...
        ....
        return queue.add(...)
            .catch(e => {
                 console.log(e);
                 // propogate errors!
                 throw e;
             })

}

La fonction de scan est une fonction asynchrone. Dans votre queue.process()fonction, vous devez attendre la fonction de numérisation, puis appeler le done()rappel.

export async function scan(job) {
  // it does some calculations, then it creates a new schedule.
  return scheduleScan(stock, period, milliseconds, "scan.js");
}

queue.process(1, (job, done) => {
  job.progress(100).then(async() => {
    await scan(job);
    done();
  });
});

export async function scheduleScan(stock, period, milliseconds, triggeredBy) {
    let uniqueId = stringHash(stock + ":" + period);
    try {
      const existingJob = await queue.getJob(uniqueId);
      if (!existingJob) {
        const job = await addJob({
          queue,
          stock,
          period,
          uniqueId,
          milliseconds,
          triggeredBy
        });
        return job;
      } else {
        const jobState = await existingJob.getState();
        if (jobState === "completed") {
          await existingJob.remove();
          const newJob = await addJob({
            queue,
            stock,
            period,
            uniqueId,
            milliseconds,
            triggeredBy
          });
          return newJob;
        }
      }
    } catch (err) {
      throw new Error(err);
    }
}

export function addJob({ queue, stock, period, milliseconds, triggeredBy }) {
  if (milliseconds) {
    return queue.add(
      { stock, period, triggeredBy },
      { delay: milliseconds, jobId: uniqueId }
    );
  } else {
    return queue.add({ stock, period, triggeredBy }, { jobId: uniqueId });
  }
}

Essaye ça! J'ai essayé de refactoriser un peu le code en utilisant async-wait.