funprog/dist/funprog.mjs

const not = async (x) => {
     var operand = await x;
    return !operand
 };

const identity = x => x;

// predicates
const isEven = x => x % 2 === 0;
const isGreaterThan = x => y => y > x ? y : 0;
const digitize = x => x ? 1 : 0;

const modulus = val => x => x % val;

const select = property => x => x[property];

function apply (x, f) {
    return f(x)
}

function compose (...funcs) {
    return x => funcs.reduceRight(apply, x)
}

function concat (accumlator, val) {
    return accumlator.concat(val)
}

function latest (accumlator, val) {
    return val
}

async function transduceAsyncIterator (transform, reducerfunction, init, asynciterator) {
    var reducer = transform(reducerfunction);
    var n = null;
    do {
        n = await asynciterator.next();
        if (!n.done) {
        var v = n.value;
        init = await reducer(init, v);
        }
    } while (!n.done)
    return init
}

async function transduceAsyncHasNextIterator (transform, reducerfunction, init, asynchasnextiterator) {
    var reducer = transform(reducerfunction);
    var n = null;
    var r = null;
    do {
        n = await asynchasnextiterator.hasNext();
        if (n) {
            var v = await asynchasnextiterator.next();
            r = await reducer(init, v);
            if (r.hasOwnProperty('reduced')) {
                if (r.reduced) {
                    init = r.reduced;
                }
            } else if (r) {
                init = r;
            }
        }
    } while (r && n)
    return init
}

async function transduceArray (xf, rf, init, xs) {
    // call reduce on the data structure internally (abstract it away)
    // pass the rf to the composed transformation
    // pass in the initial value
    var xrf = await xf(rf);
    return xs.reduce(xrf, init)
}

async function * transduceGenerator (transform, reducerfunction, init, streamgenerator) {
    var reducer = transform(reducerfunction);
    for await (const value of streamgenerator) {
        var newinit = await reducer(init, value);
        // Here we checked if there is new a 'reduced' value and only generate a new value when this is the case
        if (!newinit) ; else if (newinit === init) ; else {
            if (newinit.hasOwnProperty('reduced')) {
                if (Array.isArray(newinit.reduced)) {
                    for (var i = 0; i < newinit.reduced.length; i++) {
                        init = newinit.reduced[i];
                        yield newinit.reduced[i];
                    }
                }
            } else {
                init = newinit;
                yield newinit;
            }
        }
    }
}

/**
 * Passes all events
 */
function passthrough () {
  return function (rf) {
    // this takes 2 things and makes them 1
    return async (acc, val) => {
      return rf(acc, val) // <-- rf replaces 'concat'
    }
  }
}

/**
 * Performs a mapping operation on each record in the stream
 * @param {*} f
 */function mapping (f) {
    return function (rf) {
      // this takes 2 things and makes them 1
      return async (acc, val) => {
        var m = await f(val);
        var r = await rf(acc, m); // <-- rf replaces 'concat'
        if (r.hasOwnProperty('reduced')) {
          if (r.reduced) {
            return r
          } else {
            return acc
          }
        } else {
          return r
        }
      }
    }
  }

  /**
   * Removes records from the stream if the dont match the predicate
   * @param {*} p
   */
  function filtering (p) {
    return function (rf) {
      // this takes 2 things and makes them 1
      return async (acc, val) => {
        var pred = (await p(val));
        return pred ? rf(acc, val) : { reduced: null } // <-- rf replaces 'concat'
      }
    }
  }

  /**
   * Takes toTake records from the stream
   * @param {*} toTake
   */
function take (toTake) {
  // assert(toTake && cnt > 0)
  var count = toTake;
  return function (rf) {
    // this takes 2 things and makes them 1
    return async (acc, val) => {
      if ((--count) < 0) {
        return {
          reduced: null
        }
      } else {
        return rf(acc, val)
      }
    }
  }
}

/**
 * Skips forward in a stream by the toSkip records
 * @param {*} toSkip
 */
function skip (toSkip) {
  // assertMod(toSkip && toSkip >= 0)
  var count = 0;
  return function (rf) {
    // this takes 2 things and makes them 1
    return async (acc, val) => {
      if (count++ < toSkip) {
        return { reduced: null }
      } else {
        return rf(acc, val)
      }
    }
  }
}

/**
 * Samples from a stream at a particular frequency.
 * sample(1000) will sample a value once every second
 * @param {*} period
 */
function sampling (period) {
  // assertMod(cnt && cnt >= 0)
  var last = 0;
  return function (rf) {
    return async (acc, val) => {
      var nw = Date.now();
      var diff = nw - last;
      // console.log(diff)
      if (diff < period) {
        // console.log('Skip' + val)
        return { reduced: null }
      } else {
        last = nw;
        // console.log('Accept' + val)
        return rf(acc, val)
      }
    }
  }
}

/**
 * Transforms a series of measurements into events.  The predicate determines when the measurements continue the event or non-event
 * @param {*} predicate
 */
function eventing (predicate) {
  var latest = null;
  var sequence = -1;
  return function (rf) {
    return async (acc, val) => {
      sequence++;
      var pred = (await predicate(val));
      if (pred && !latest) {
        latest = {
          start: val.time ? val.time : sequence,
          end: val.time ? val.time : sequence
        };
        return { reduced: null }
      } else if (pred && latest) {
        latest.end = val.time ? val.time : sequence;
        return { reduced: null }
      } else if (!pred && latest) {
        var next = latest;
        latest = null;
        return rf(acc, next)
      } else {
        return { reduced: null }
      }
    }
  }
}

/**
 *  Split one event into multiple
 *  splitter is a function which maps a value to an array
 */
function split (splitter) {
  return function (rf) {
    // this takes 2 things and makes them 1
    return async (acc, val) => {
      var rs = await splitter(val);
      var reduction = { reduced: [] };
      try {
        var acc2 = acc;
        for (var i = 0; i < rs.length; i++) {
          var r = rs[i];
          acc2 = await rf(acc2, r);
          if (acc2.hasOwnProperty('reduced')) {
            if (acc2.reduced) {
              reduction.reduced.push(acc2.reduced);
            }
          } else {
            reduction.reduced.push(acc2);
          }
        }
        return reduction
      } catch (ex) {
        console.log(ex);
      }
    }
  }
}

/**
 * Randomly samples from a stream with a target frequency.
 * randomFilter(100) will target to sample one value in every 100
 * @param {*} countFrequency - a number
 */
function randomFilter (countFrequency) {
  // assertMod(cnt && cnt >= 0
  return function (rf) {
    return async (acc, val) => {
      var rand = Math.floor((Math.random() * countFrequency));
      if (rand === 0) {
        return rf(acc, val)
      } else {
        return { reduced: null }
      }
    }
  }
}

 /**
   * Make n neighbors available as each stream record is processed.
   * @param {*} numberOfNeighbors
   */
  function neighbors (numNeighbors = 10) {
    var neighbors = [];

    return function (rf) {
      return async (acc, val) => {
        neighbors.push(val);
        if ((neighbors.length) < numNeighbors) {
          return {
            reduced: null
          }
        } else {
          let first = neighbors[0];
          let enriched = Object.assign({}, first);
          enriched.neighbors = neighbors;
          enriched.data = first;
          let res = rf(acc, enriched);
          neighbors.shift();
          return res
        }
      }
    }
  }

async function * makeAsyncRangeIterator (start = 0, end = Infinity, step = 1, delay = 10) {
    let iterationCount = 0;
    for (let i = start; i < end; i += step) {
        iterationCount++;
        yield new Promise((resolve, reject) => {
            setTimeout(() => {
                resolve(i);
            }, delay);
        });
    }
    return iterationCount
  }

function makeAsyncHasNextRangeIterator (start = 0, end = Infinity, step = 1) {
    let iterationCount = start;
    return {
        hasNext: function () {
            return iterationCount < end
        },
        next: function () {
            return new Promise((resolve, reject) => {
                setTimeout(() => {
                    resolve(iterationCount++);
                }, 10);
            })
        }
    }
}

function makeArrayIterator (dataArray) {
    let iterationCount = 0;
    return {
        hasNext: function () {
            return iterationCount < dataArray.length
        },
        next: function () {
            return dataArray[iterationCount++]
            }
        }
    }

export { not, identity, isEven, isGreaterThan, digitize, modulus, select, compose, concat, latest, transduceAsyncIterator, transduceAsyncHasNextIterator, transduceArray, transduceGenerator, mapping, filtering, take, skip, eventing, sampling, passthrough, split, randomFilter, neighbors, makeArrayIterator, makeAsyncRangeIterator, makeAsyncHasNextRangeIterator };
//# sourceMappingURL=funprog.mjs.map