pipe

*pipe.md

Ever changing implementations of pipes that I reimplement in each codebase for different use cases.

If you do want a package instead. You should go with this https://github.com/barelyhuman/pipe

active-pipe.js

/**
 *
 * @param  {...function} args
 * @returns
 * @example
 * console.log(
 *   await pipe(
 *     () => [1, 2, 3],
 *     () => pipe(() => prev.map((x) => x * 2)),
 *     (counts) => counts.map(async (x) => x * 2)
 *   )
 * );
 */
function pipe(...args) {
  return args.reduce((acc, item, index) => {
    if (typeof item !== "function")
      throw new Error(`[pipe] item at index: ${index} is not a function`);
    return acc.then((prev) => {
      const res = item(prev);
      return Array.isArray(res) && res.every((x) => x instanceof Promise)
        ? Promise.all(res)
        : res;
    });
  }, Promise.resolve());
}

pipe.js

const pipe = (initData, chain = []) => {
  let _data = initData;

  if (typeof initData === "function") {
    _data = initData();
  }

  return {
    _(fn) {
      chain.push(fn);
      return this;
    },
    async value() {
      let initValue = _data;
      if (_data instanceof Promise) {
        initValue = await _data;
      }
      return chain.reduce((acc, item) => {
        return acc.then((prev) => {
          return item(prev);
        });
      }, Promise.resolve(initValue));
    },
  };
};

sync-async-active-pipe.js

const isArrayOfPromises = (item) =>
  Boolean(Array.isArray(item) && item.every((x) => x instanceof Promise));

const resolveArray = (item) =>
  (isArrayOfPromises(item) && Promise.all(item)) || item;


/**
 * @description handles both sync and async cases. If even on function in the
 * pipe params is async or returns an array of promises, the entire pipe is considered
 * to be async.
 * if not then it runs the functions in sync
 * @param  {...function} args
 * @returns
 * @example
 * // gives the result as sync, instead of a promise.
 * const nextHour = pipe(
 *   () => new Date(),
 *   (d) => d.setHours(d.getHours() + 1),
 *   (x) => new Date(x)
 * )
 *
 * // gives the result as async, and needs to be awaited
 * const nextHour = await pipe(
 *   () => new Date(),
 *   async (d) => d.setHours(d.getHours() + 1),
 *   (x) => new Date(x)
 * )
 */
export const pipe = (...args) => {
  let syncFlow = true,
    result = undefined;

  for (let i = 0; i < args.length; i += 1) {
    const fn = args[i];
    if (!syncFlow) {
      result = Promise.resolve(result)
        .then(resolveArray)
        .then((d) => fn(d));
    } else {
      result = fn(result);
      if (result instanceof Promise || isArrayOfPromises(result)) {
        syncFlow = false;
      }
    }
  }

  if (result && typeof result.then === "function") {
    return result.then(resolveArray);
  }

  return result;
};