refactor: build plugin pipeline parameters at initialization

In addition, factorize the pipeline config function to avoid code duplication.

lib/plugins/index.js

@@ -1,4 +1,4 @@
-const {isPlainObject, omit, castArray, isUndefined} = require('lodash');
+const {identity, isPlainObject, omit, castArray, isUndefined} = require('lodash');
 const AggregateError = require('aggregate-error');
 const getError = require('../get-error');
 const PLUGINS_DEFINITIONS = require('../definitions/plugins');
@@ -7,31 +7,37 @@ const normalize = require('./normalize');
 
 module.exports = (options, pluginsPath, logger) => {
   const errors = [];
-  const plugins = Object.entries(PLUGINS_DEFINITIONS).reduce((plugins, [type, {configValidator, default: def}]) => {
-    let pluginConfs;
+  const plugins = Object.entries(PLUGINS_DEFINITIONS).reduce(
+    (
+      plugins,
+      [type, {configValidator, default: def, pipelineConfig, postprocess = identity, preprocess = identity}]
+    ) => {
+      let pluginConfs;
 
-    if (isUndefined(options[type])) {
-      pluginConfs = def;
-    } else {
-      // If an object is passed and the path is missing, set the default one for single plugins
-      if (isPlainObject(options[type]) && !options[type].path && castArray(def).length === 1) {
-        options[type].path = def;
+      if (isUndefined(options[type])) {
+        pluginConfs = def;
+      } else {
+        // If an object is passed and the path is missing, set the default one for single plugins
+        if (isPlainObject(options[type]) && !options[type].path && castArray(def).length === 1) {
+          options[type].path = def;
+        }
+        if (configValidator && !configValidator(options[type])) {
+          errors.push(getError('EPLUGINCONF', {type, pluginConf: options[type]}));
+          return plugins;
+        }
+        pluginConfs = options[type];
       }
-      if (configValidator && !configValidator(options[type])) {
-        errors.push(getError('EPLUGINCONF', {type, pluginConf: options[type]}));
-        return plugins;
-      }
-      pluginConfs = options[type];
-    }
 
-    const globalOpts = omit(options, Object.keys(PLUGINS_DEFINITIONS));
+      const globalOpts = omit(options, Object.keys(PLUGINS_DEFINITIONS));
+      const steps = castArray(pluginConfs).map(conf => normalize(type, pluginsPath, globalOpts, conf, logger));
 
-    plugins[type] = pipeline(
-      castArray(pluginConfs).map(conf => normalize(type, pluginsPath, globalOpts, conf, logger))
-    );
+      plugins[type] = async input =>
+        postprocess(await pipeline(steps, pipelineConfig && pipelineConfig(plugins, logger))(await preprocess(input)));
 
-    return plugins;
-  }, {});
+      return plugins;
+    },
+    {}
+  );
   if (errors.length > 0) {
     throw new AggregateError(errors);
   }

lib/plugins/normalize.js

@@ -42,6 +42,7 @@ module.exports = (type, pluginsPath, globalOpts, pluginOpts, logger) => {
   const validator = async input => {
     const {outputValidator} = PLUGINS_DEFINITIONS[type] || {};
     try {
+      logger.log('Call plugin "%s"', type);
       const result = await func(cloneDeep(input));
       if (outputValidator && !outputValidator(result)) {
         throw getError(`E${type.toUpperCase()}OUTPUT`, {result, pluginName});

lib/plugins/pipeline.js

@@ -8,10 +8,6 @@ const {extractErrors} = require('../utils');
  *
  * @typedef {Function} Pipeline
  * @param {Any} input Argument to pass to the first step in the pipeline.
- * @param {Object} options Pipeline options.
- * @param {Boolean} [options.settleAll=false] If `true` all the steps in the pipeline are executed, even if one rejects, if `false` the execution stops after a steps rejects.
- * @param {Function} [options.getNextInput=identity] Function called after each step is executed, with the last step input and the current current step result; the returned value will be used as the input of the next step.
- * @param {Function} [options.transform=identity] Function called after each step is executed, with the current step result and the step function; the returned value will be saved in the pipeline results.
  *
  * @return {Array<*>|*} An Array with the result of each step in the pipeline; if there is only 1 step in the pipeline, the result of this step is returned directly.
  *
@@ -22,9 +18,14 @@ const {extractErrors} = require('../utils');
  * Create a Pipeline with a list of Functions.
  *
  * @param {Array<Function>} steps The list of Function to execute.
+ * @param {Object} options Pipeline options.
+ * @param {Boolean} [options.settleAll=false] If `true` all the steps in the pipeline are executed, even if one rejects, if `false` the execution stops after a steps rejects.
+ * @param {Function} [options.getNextInput=identity] Function called after each step is executed, with the last step input and the current current step result; the returned value will be used as the input of the next step.
+ * @param {Function} [options.transform=identity] Function called after each step is executed, with the current step result, the step function and the last step input; the returned value will be saved in the pipeline results.
+ *
  * @return {Pipeline} A Function that execute the `steps` sequencially
  */
-module.exports = steps => async (input, {settleAll = false, getNextInput = identity, transform = identity} = {}) => {
+module.exports = (steps, {settleAll = false, getNextInput = identity, transform = identity} = {}) => async input => {
   const results = [];
   const errors = [];
   await pReduce(
@@ -33,7 +34,7 @@ module.exports = steps => async (input, {settleAll = false, getNextInput = ident
       let result;
       try {
         // Call the step with the input computed at the end of the previous iteration and save intermediary result
-        result = await transform(await step(lastInput), step);
+        result = await transform(await step(lastInput), step, lastInput);
         results.push(result);
       } catch (err) {
         if (settleAll) {