[Node.js] Construire un gestionnaire de processus propre

Dans cet article, je vais démontrer comment construire un gestionnaire de processus simple et maintenable pour Node.js, en tirant parti de son Event Loop.

L’idée principale est d’avoir un processeur central capable d’exécuter une série de tâches, de manière synchrone. Comme vous le savez peut-être, JavaScript est par essence un langage asynchrone. Prenons un exemple simple :

1
2
3
4
5
6
7
8
function CrawlingAWebPageAndGetLinks(){
  // do the stuff the method pretend;
}

(function program(){
  var result = CrawlingAWebPageAndGetLinks();
  console.log(result);
})();

Comme le flux d’exécution est asynchrone, la variable result ne sera pas valorisée avant que la méthode de crawling ne se termine, et l’utilisation de la variable se produira avant sa valorisation. La solution standard en JavaScript pour gérer ce problème est d’utiliser une méthode callback :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
function CrawlingAWebPageAndGetLinks(callback){
  // do the stuff the method pretend;
  var value = Crawler.Get("https://www.google.com/search?q=callback%20hell");
  // then use
  callback(value);
}

(function program(){
  var callback = function(v){
    console.log(v);
  };
  var result = CrawlingAWebPageAndGetLinks(callback);
})();

Les callbacks sont vraiment pratiques, mais que se passe-t-il si j’ai un processus avec plusieurs tâches ?

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
var results = {};
function checkInDatabase(callback){
  results.checkresult = SomeComputationThatTakesTime();
  callback();
}
function crawl(callback){
  results.crawlingResult = SomeComputationThatTakesTime();
  callback();
}
function extractData(callback){
  results.data = SomeComputationThatTakesTime();
  callback();
}
function saveData(callback){
  results.dbAcknowledgementMessage = SomeComputationThatTakesTime();
  callback();
}
function displayProcessResult(){
  console.log(results);
}

function process (){
  checkInDatabase(function(){
    crawl(function(){
      extractData(function(){
        saveData(function(){
          displayProcessResult();
        });
      });
    });
  });
}

Oui, nous venons de tomber dans le Callback Hell ! Le principal problème avec les callbacks imbriqués est que votre code manque de lisibilité et de flexibilité. Il est assez difficile de comprendre ce que le code est censé faire. Et si vous devez modifier le processus, vous devrez repenser l’ensemble du processus.

Nous pouvons faire bien mieux !

Le modèle de traitement de NodeJs est basé sur une event loop, donc nous pouvons facilement utiliser une approche événementielle. Nous construisons un “event bus” très simple :

1
2
3
4
5
6
7
var processEvents = {
    taskDone: 'taskDone',
    taskFaulted: 'taskFaulted',
    processDone: 'processDone',    
};

module.exports = processEvents;

Une tâche est une fonction qui encapsule une unité logique de travail et un callback vers l’event bus :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
var Merger = require("../../merger.js");

function mergeCrawlerResults() {
    this.execute = function(context, eventBus) {
    	var taskCompleteCallback = function(data) {
            context.crawlerMergeResults = data;
            eventBus.taskDone();
        }
        
        var merger = new Merger(taskCompleteCallback);
        merger.merge(context.crawledNewLinks);
    }
}

module.exports = mergeCrawlerResults;

Notez que j’utilise un objet context, où une tâche peut stocker des données que les tâches suivantes peuvent utiliser, et utiliser des données précédemment définies.

Ensuite, nous pouvons construire le processeur. Comme nous voulons le maximum de lisibilité, nous voulons que notre processus soit une liste déclarative de tâches. Donc notre processeur exécute les tâches de manière synchrone, attendant un événement de l’event bus avant d’exécuter la tâche suivante :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
var processEvents = require("./processEvents.js"),
    ProcessEventBus = require("./processManager/processEventBus.js");

function Processor(process, onCompletion, options) {
    options = options || {};
    options.logEvents = options.hasOwnProperty('logEvents') ? options.logEvents : false;
    options.context = options.hasOwnProperty('context') ? options.context : {};

    var eventBus = new ProcessEventBus();

    function executeNext() {
        if(options.logEvents) console.log('Processor.executeNext, tasks left : ' + process.length);
        
        if (process.length > 0) {
            var nextTask = process.shift();
            nextTask.execute(options.context, eventBus);
        }
        else
            eventBus.processDone();
    }

    eventBus.on(processEvents.taskDone, function(data) {
        if(options.logEvents) console.log('processEvents.taskDone');
        executeNext();
    });
    
    eventBus.on(processEvents.processDone, function(data) {
        if(options.logEvents) console.log('processEvents.processDone');
        onCompletion(context);
    });
    
    this.execute = function() {
        if(options.logEvents) console.log('Processor.execute, initial tasks count : ' + process.length);
        executeNext();
    };
}

module.exports = Processor;

Puis, la définition du processus lui-même : Nous rassemblons d’abord tous les composants nécessaires, puis construisons simplement un tableau des tâches dans l’ordre requis :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
var // infra & services
    Store = require("../../dal.js"),
    // process infrastructure
    Processor = require("../processManager/processor.js"),
    // tasks 
    GetStoredActiveLinks = require("./tasks/getStoredActiveLinks.js"),
    GetAllLinks = require("./tasks/getAllLinks.js"),
    Crawl = require("./tasks/crawl.js"),
    MergeCrawlerResults = require("./tasks/mergeCrawlerResults.js"),
    CheckForDelta = require("./tasks/checkForDelta.js"),
    SaveNewLinks = require("./tasks/saveNewLinks.js"),
    SaveLinksSummary = require("./tasks/saveLinksSummary.js");

function Process() {
    
    var store = new Store();
    
    var process = [
          new GetStoredActiveLinks(store)
        , new GetAllLinks(store)
        , new Crawl()
        , new MergeCrawlerResults()
        , new CheckForDelta()
        , new SaveNewLinks(store)
        , new SaveLinksSummary(store)
    ];

    this.execute = function() {
        console.log('new process');
        var onProcessComplete = function(result){ console.log("process completed")};
        var processor = new Processor(process, onProcessComplete);
        processor.execute();
    }
};

module.exports = Process;

C’est fait ! Nous avons maintenant un processeur de tâches joliment découplé, et nous pouvons maintenant nous concentrer sur l’écriture d’unités de travail qui seront encapsulées dans des tâches, donc nous serons maintenant capables de suivre le Single Responsibility Principle.

Commentaires