Événements personnalisés avec des processus enfants en Node.js

Question

J'utilise actuellement un processus enfant afin d'exécuter certaines fonctions de calcul de manière non bloquante.

Je me heurte à un problème : je dois pouvoir relier le résultat d'un calcul au callback correct dans le processus parent. Par exemple :

var cp = require('child_process');
var n = cp.fork(__dirname + '/child.js');

exports.evaluate = function(data, callback) {

    n.send({ data : data});

    n.once('message', function(result) {

        callback(result);

    }

}

Le problème avec cet exemple est que si un calcul revient avant un autre, il ne recevra pas les bons résultats.

Existe-t-il un moyen d'utiliser un nom d'événement personnalisé au lieu de 'message' afin de s'assurer que chaque fois que la fonction d'évaluation est appelée, elle crée un écouteur unique qui sera supprimé une fois qu'elle aura été appelée ?

Comment puis-je émettre un événement personnalisé à partir d'un processus enfant ?

Answer 1

Créez un nouvel EventEmitter avec new events.EventEmitter . Puis émettre un événement basé sur tout message avec eventType:"someType" vous recevez.

c'est-à-dire quelque chose comme

 events = require("events")     
 emitter = new events.EventEmitter
 n.on("message", function(msg) { emitter.emit(msg.eventType,msg.body) })

Ensuite, liez les écouteurs (avec once o on ) à cet émetteur d'événements.

Answer 2

Je pense qu'une bonne solution serait d'appeler votre fonction evaluate de manière récursive jusqu'à ce que tous les calculs soient terminés. Utilisez une file d'attente pour vos calculs et traitez-les de manière FIFO.

var computations = [];

Vous devriez abandonner le n.once('message' et, en dehors de votre fonction evalutate, définir un gestionnaire n.on('message'. Pour chaque "message" que vous recevez et qui correspond à votre type de données d'évaluation (appelons-le "résultat"). Ensuite, vous enregistrez ce résultat dans une file FIFO de résultats. Après quoi, on vérifie s'il y a d'autres calculs et on appelle à nouveau evaluate. Si vous le souhaitez, vous pouvez nettoyer ce processus pour séparer le calcul le plus ancien dans la fonction d'évaluation.

var results = [];

n.on('message', function(m) {
  if(m.msg === "result") { // message looks like {msg: "result", data: 1234}
    results.push(m.data);
  }
  if(computations.length > 0) {
    var comp = computations[0]; // Save oldest computation
    computations.splice(0,1); // Remove oldest computation from array
    evaluate(comp); // Evaluate oldest computation
  }
  else
    process_results(); // if you have 0 computations left, 
                       // you got the last result so go process them.
});

Avant l'appel de evaluate, vous devez pousser votre calcul dans la file d'attente, et envoyer le plus ancien à evaluate.

if(computations.length > 0) {
  computations.push(new_comp); // Push on your newest compuation
  var comp = computations[0]; // Save oldest computation
  computations.splice(0,1); // Remove oldest computation from array
  evaluate(comp); // Evaluate oldest computation
}
else {
  evaluate(new_comp); // Evaluate computation
}
evaluate(new_comp); // Evaluate oldest computation

Votre drapeau "DONE" sera activé lorsque computations.length === 0. Et cela se fera naturellement avec le code ci-dessus. Chaque fois qu'il vous reste 0 calcul dans le handler n.on('message',..) vous pouvez appeler votre fonction de traitement des résultats :

Pour traiter vos résultats, il s'agit d'une simple boucle for.

function process_results() {
    if(results.length > 0)
      for(x in results)
        console.log("New result: "+x);
}

Answer 3

Ce n'est peut-être pas tout à fait "correct" et il y a plus de six ans, mais cela semble fonctionner en 2020.

> process.version
'v12.14.1'
> var ignore = process.on('custom',function(e){console.dir(e,{depth:null});})
undefined
> process.emit('custom',{e:"hello"})
{ e: 'hello' }
true
> 

ie tout objet passé dans un process.emit(x,obj) est envoyé à ce qui est enregistré par rapport à une correspondance process.on(x,function(obj){ /* ... */ }) rappel