Traiter les tâches avec un worker

Si votre API génère elle-même les factures, une opération lente immobilise les ressources du serveur qui reçoit les commandes. Vous devez pouvoir faire tourner le traitement à part, sans que le worker prenne en charge les requêtes HTTP. Cette séparation facilite aussi l’ajustement du nombre de traitements actifs selon la charge.

La leçon précédente a établi que l’API persiste une commande avant de publier sa tâche. Ici, vous construisez le côté consommateur : il reçoit les messages, vérifie leur format puis demande au service de générer la facture. Le worker reçoit ses dépendances de la composition de l’application, comme le routeur et le producteur les recevaient dans la leçon sur la publication.

Un consommateur indépendant et injectable

Le contrat du projet ne prévoit pas de fichier d’entrée de worker ni de bibliothèque de file particulière. Pour ne pas inventer un nom de fichier ou l’API d’un fournisseur, le module ci-dessous expose une fonction de démarrage à appeler depuis la composition de déploiement déjà existante. Le consommateur injecté doit fournir consume(handler, { concurrency }) et close() ; cette interface permet de brancher le fournisseur réellement choisi sans coupler le traitement à son SDK.

function creerTacheFacture(commandeId) {
  return {
    type: 'facture.generer',
    version: 1,
    commandeId
  };
}

function validerTacheFacture(tache) {
  return Boolean(
    tache &&
    tache.type === 'facture.generer' &&
    tache.version === 1 &&
    typeof tache.commandeId === 'string' &&
    tache.commandeId.trim().length > 0
  );
}

function creerWorkerFactures({ consommateur, factures, journal, concurrence }) {
  if (!consommateur || typeof consommateur.consume !== 'function') {
    throw new TypeError('Un consommateur de tâches est requis.');
  }
  if (!factures || typeof factures.generer !== 'function') {
    throw new TypeError('Un service de facturation est requis.');
  }
  if (!journal || typeof journal.info !== 'function' || typeof journal.error !== 'function') {
    throw new TypeError('Un journal compatible est requis.');
  }
  if (!Number.isInteger(concurrence) || concurrence < 1) {
    throw new TypeError('La concurrence doit être un entier positif.');
  }

  let arrete = false;
  let fermetureEnCours;

  async function traiter(tache) {
    if (!validerTacheFacture(tache)) {
      throw new Error('Format de tâche de facture invalide.');
    }

    await factures.generer(tache.commandeId);
    journal.info({
      evenement: 'facture.generee',
      commandeId: tache.commandeId
    });
  }

  async function demarrer() {
    journal.info({ evenement: 'worker.factures.demarrage', concurrence });
    await consommateur.consume(traiter, { concurrency: concurrence });
  }

  async function arreterProprement(signal) {
    if (fermetureEnCours) {
      return fermetureEnCours;
    }

    arrete = true;
    journal.info({ evenement: 'worker.factures.arret', signal });
    fermetureEnCours = consommateur.close();
    await fermetureEnCours;
    journal.info({ evenement: 'worker.factures.arrete' });
  }

  function installerSignaux(processus = process) {
    const gererSignal = (signal) => {
      arreterProprement(signal).catch((erreur) => {
        journal.error({
          evenement: 'worker.factures.erreur_arret',
          message: erreur.message
        });
        processus.exitCode = 1;
      });
    };

    processus.once('SIGTERM', gererSignal);
    processus.once('SIGINT', gererSignal);

    return () => {
      processus.removeListener('SIGTERM', gererSignal);
      processus.removeListener('SIGINT', gererSignal);
    };
  }

  return {
    demarrer,
    arreterProprement,
    installerSignaux,
    estArrete: () => arrete
  };
}

module.exports = {
  creerTacheFacture,
  validerTacheFacture,
  creerWorkerFactures
};

Le gestionnaire transmet uniquement l’identifiant métier au service de facturation, après validation du message. Si le traitement échoue, l’erreur remonte au consommateur : c’est à l’adaptateur de file de ne confirmer le message qu’après une exécution réussie et de conserver le message en cas d’échec. La stratégie de nouvelles tentatives et de reprise sera définie dans la leçon dédiée à la résistance aux échecs.

Régler la concurrence et arrêter proprement

La concurrence indique combien de tâches le consommateur peut traiter simultanément. La valeur doit venir de la configuration de l’environnement et être fournie à la fonction, plutôt que d’être inscrite en dur dans le code. Commencez par une valeur compatible avec les ressources disponibles et la capacité de la base ou du service externe appelé par la génération de factures.

La fonction installerSignaux() permet à la composition du worker de relier l’arrêt gracieux aux signaux d’arrêt usuels d’un processus Node.js. Après avoir créé le consommateur, le service de facturation et le journal, la composition appelle cette fonction, puis demarrer(). Au déploiement, lancez cette composition comme un processus distinct du serveur web ; le contrat fourni ne nommant aucun fichier d’entrée, le point de lancement reste celui défini par votre environnement de déploiement.

À retenir

• Le worker consomme les messages sans recevoir les requêtes HTTP de l’API.
• Injectez le consommateur, le service de facturation, le journal et la concurrence.
• Validez le format avant d’appeler la génération de facture.
• L’arrêt gracieux doit cesser les nouvelles livraisons et attendre la fin des traitements actifs.

La prochaine leçon rendra le traitement résistant aux échecs, notamment en définissant les règles de reprise des tâches qui n’aboutissent pas.