funprog/dist/funprog.umd.js

(function (global, factory) {
    typeof exports === 'object' && typeof module !== 'undefined' ? factory(exports) :
    typeof define === 'function' && define.amd ? define(['exports'], factory) :
    (global = global || self, factory(global.funprog = {}));
}(this, function (exports) { 'use strict';

    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++]
                }
            }
        }

    exports.not = not;
    exports.identity = identity;
    exports.isEven = isEven;
    exports.isGreaterThan = isGreaterThan;
    exports.digitize = digitize;
    exports.modulus = modulus;
    exports.select = select;
    exports.compose = compose;
    exports.concat = concat;
    exports.latest = latest;
    exports.transduceAsyncIterator = transduceAsyncIterator;
    exports.transduceAsyncHasNextIterator = transduceAsyncHasNextIterator;
    exports.transduceArray = transduceArray;
    exports.transduceGenerator = transduceGenerator;
    exports.mapping = mapping;
    exports.filtering = filtering;
    exports.take = take;
    exports.skip = skip;
    exports.eventing = eventing;
    exports.sampling = sampling;
    exports.passthrough = passthrough;
    exports.split = split;
    exports.randomFilter = randomFilter;
    exports.neighbors = neighbors;
    exports.makeArrayIterator = makeArrayIterator;
    exports.makeAsyncRangeIterator = makeAsyncRangeIterator;
    exports.makeAsyncHasNextRangeIterator = makeAsyncHasNextRangeIterator;

    Object.defineProperty(exports, '__esModule', { value: true });

}));
//# sourceMappingURL=funprog.umd.js.map