Let's pipe!

A common task for a developer is to perform transformations of data in several steps. One could implement it with several assignments

export function transform(array) {
  const incrementedBy2 = incrementBy2(array);
  const doubled = double(incrementedBy2);
  const total = sum(doubled);
  return total;
}

or without assigning each step result to a new variable.

function transformNoAssignments(array) {
  return sum(double(incrementBy2(array)));
}

None of these solutions works perfectly. transform() is quite tedious to write and adds intermediate assignments that we do not really care about.

Even though transformNoAssignments() is shorter, it still has issues from a readability perspective. For one, the steps are run in reverse order of how they are written. IncrementBy2() would run first, then double() and at last sum().

In this blog post we will implement a function, pipe(), that addresses the aforementioned issues.

let numbers = [1, 2, 3];
let transformed = pipe(incrementBy2, double, sum)(numbers);

Pretty, right? And readable, none the less.

Implementation

Going forward we call the set of input functions as pipeline, while each individual function in our pipeline will be referred to as a step.

The implementation of pipe() is shown below, both with traditional function syntax

function pipe(...steps) {
  return function (initialValue) {
    return steps.reduce(
            function(accumulatedValue, currentFunction){ 
                return currentFunction(accumulatedValue)
            }, 
            initialValue)
    };
}

... and arrow function syntax.

const pipe = (...steps) => 
    (initialValue) =>
          steps.reduce(
              (accumulatedValue, currentFunction) =>
                    currentFunction(accumulatedValue), 
              initialValue);

pipe() is a function that accepts any number of functions as input parameters and returns a function that accepts an initialValue. When executed, the steps are processed sequentially, with the output of the previous step serving as input to the next step in the pipeline. InitialValue is used as input to the first step.

Let us run through the details.

How can the returned function accept any number of functions as input?

A JavaScript function can be called with any number of arguments. Should the number of called parameters exceed the number of input parameters those excess inputs would be ignored.

const add = (a, b) => a + b;
add(1, 2) // 3
add(1, 2, 3, 4) // 3

However, when using the spread operator (...) the inputs are converted into a single array.

const logNumbersAsArray = (...numbers) => console.log(numbers);

logNumbersAsArray(1,2,3,4,5) // [1,2,3,4,5]

Consequently, we can call pipe() with as many input parameters as we like. The steps would be converted into an array of functions.

Next on, we look at how Reduce() is used

Reduce() is a JavaScript function that operates on an array, executes a reducer function on each element, ultimately reducing the array to a single output value.

One could for instance implement a sum function as follows

const sumReducer = (accumulated, current) => accumulated + current;

const sum = (numbers) => numbers.reduce(sumReducer, 0);

Utilizing logging statements we can inspect the intermediate steps of the process.

const sum = (numbers) => 
  numbers.reduce(
        (accumulated, current) => {
    const result = accumulated + current;
    console.log(`Accumulated: ${accumulated} Current: ${current} Output: ${result}`);
            return result;
        }, 0)

const result = sum([1,2,3,4]);

// Accumulated: 0. Current: 1. Output: 1
// Accumulated: 1. Current: 2. Output: 3
// Accumulated: 3. Current: 3. Output: 6
// Accumulated: 6. Current: 4. Output: 10

console.log(result);
// 10

Since pipe() accepts functions instead of numbers as input we would have to call each step as part of the reducer function. Take note of how currentFunction() is called with accumulatedValue as input.

const pipe = (...steps) => 
    (initialValue) =>
          steps.reduce(
              (accumulatedValue, currentFunction) =>
                    currentFunction(accumulatedValue), 
              initialValue);

Verification

To verify that the implementation works as expected we write some Jest tests. We test that each step is passed the returned value of the previous step and that pipe() outputs the correct value when all steps has been processed.

const incrementBy2 = jest.fn((number) => number.map((x) => x + 2));
const double = jest.fn((number) => number.map((x) => x * 2));
const sum = jest.fn((number) => number.reduce((a, b) => a + b, 0));

describe("Pipe", () => {
  it("should pass output of current step to next", () => {
    pipe(incrementBy2, double, sum)([1, 2, 3]);

    expect(incrementBy2).toHaveBeenCalledWith([1, 2, 3]);
    expect(double).toHaveBeenCalledWith([3, 4, 5]);
    expect(sum).toHaveBeenCalledWith([6, 8, 10]);
  });

  it("should return accumulated value", () => {
    const twentyFour = 
      pipe(incrementBy2, double, sum)([1, 2, 3]);
      expect(twentyFour).toBe(24);

    const eighteen = pipe(double, incrementBy2, sum)([1, 2, 3]);
    expect(eighteen).toBe(18);
  });
});

Closing

We conclude this blog post by showing an example of how pipe() can be used.

// Steps
const incrementBy2 = (array) => array.map((x) => x + 2);
const double = (array) => array.map((x) => x * 2);
const sum = (array) => array.reduce((a, b) => a + b, 0);

let numbers = [1, 2, 3];

let pipeline = pipe(incrementBy2, double, sum);
typeof(pipeline) // function
pipeline(numbers); // 45

// called directly
pipe(incrementBy2, double, sum)(numbers); // 45
});