Added simple tap function #14

src/pipeline.js

@@ -41,6 +41,16 @@ const nopLogger = {
  * @return {Promise<Context>} Promise which resolves to a parameters to be added to the context.
 */
 
+/**
+ * Tap function
+ *
+ * @typedef {function(context, _action, index)} tapFunction
+ * @callback tapFunction
+ * @param {Context} context Pipeline execution context that is passed along
+ * @param {Action} action Pipeline action define during construction
+ * @param {number} index index of the function invocation order
+*/
+
 /**
  * Pipeline that allows to execute a list of functions in order. The pipeline consists of 3
  * major function lists: `pre`, `once` and, `post`. the functions added to the `pre` list are
@@ -67,6 +77,8 @@ class Pipeline {
     this._oncef = null;
     // functions that are executed after
     this._posts = [];
+    // functions that are executed before each step
+    this._taps = [];
   }
 
   /**
@@ -91,6 +103,18 @@ class Pipeline {
     return this;
   }
 
+  /**
+   * Adds a tap (observing) function to the pipeline. taps are executed for every
+   * single pipeline step and best used for validation, and logging. taps don't have
+   * any effect, i.e. the return value of a tap function is ignored.
+   * @param {pipelineFunction} f function to be executed in every step. Effects are ignored.
+   */
+  tap(f) {
+    this._taps.push(f);
+    this._last = this._taps;
+    return this;
+  }
+
   /**
    * Adds a condition to the previously defined `pre` or `post` function. The previously defined
    * function will only be executed if the predicate evaluates to something truthy or returns a
@@ -163,12 +187,14 @@ class Pipeline {
      * @param {pipelineFunction} currFunction Function that is currently "reduced"
      * @returns {Promise} Promise resolving to the new value of the accumulator
      */
-    const merge = (currContext, currFunction) => {
+    const merge = (currContext, currFunction, index) => {
       // copy the pipeline payload into a new object to avoid modifications
       const mergedargs = _.merge({}, currContext);
 
       // log the function that is being called and the parameters of the function
-      this._action.logger.silly('processing ', { function: `${currFunction}`, params: mergedargs });
+      this._action.logger.silly('processing ', { function: `${currFunction}`, index, params: mergedargs });
+
+      this._taps.map(f => f(mergedargs, this._action, index));
 
       return Promise.resolve(currFunction(mergedargs, this._action))
         .then((value) => {

test/testPipeline.js

@@ -219,4 +219,12 @@ describe('Testing Pipeline', () => {
       done();
     });
   });
+
+  it('Executes taps', (done) => {
+    new Pipeline({ logger })
+      .pre(() => ({ foo: 'bar' }))
+      .post(() => ({ bar: 'baz' }))
+      .tap((c, a, i) => { if (i === 1) { done(); } })
+      .run();
+  });
 });