text: /* ############################################################################
  The MIT License (MIT)
  
  Copyright (c) 2019 Van Schroeder
  Copyright (c) 2019 Webfreshener, LLC
  
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
  
  The above copyright notice and this permission notice shall be included in all
  copies or substantial portions of the Software.
  
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  SOFTWARE.
  
  ############################################################################ */
  import {_observers, Validator} from "./Validator";
  import {fill, mapArgs} from "./Utils";
  import {Executor} from "./Executor";
  import {default as DefaultVOSchema} from "../schemas/default-pipe-vo.schema";
  import {Properties} from "./Properties";
  
  const _pipes = new WeakMap();
  const _cache = new WeakMap();
  
  const clearPipeInterval = (pipe) => {
      const _ivl = _pipes.get(pipe).tO;
      if (_ivl) {
          clearInterval(_ivl);
      }
  }
  
  /**
   * Pipeline Class
   */
  export class Pipeline {
      static getExecs(..._pvs) {
          return _pvs.map((_p) => {
              _p = Array.isArray(_p) ? _p[0] : _p;
              return (d) => {
                  const _exec = ((typeof _p === "function") ? _p : void 0) ||
                      // is pipeline or implements pipeline api
                      (_p["exec"]) ||
                      // is validator or implements validator api
                      (_p["validate"] ? ((d) => _p["validate"](d) ? d : false) : void 0) ||
                      // default
                      ((_) => _);
                  return (_exec).apply(null, [d]);
              };
          });
      }
  
      /**
       *
       * @param pipesOrVOsOrSchemas
       */
      constructor(...pipesOrVOsOrSchemas) {
          _cache.set(this, []);
  
          // TODO: solve this issue with async methods to remove kludge
          if (pipesOrVOsOrSchemas[0] instanceof Function) {
              pipesOrVOsOrSchemas.splice(0, 0, DefaultVOSchema);
          }
  
          pipesOrVOsOrSchemas = mapArgs(...pipesOrVOsOrSchemas);
  
          // stores config & state
          _pipes.set(this,
              Properties.init(this, {
                  callbacks: fill(Pipeline.getExecs(...pipesOrVOsOrSchemas)),
                  pipesOrVOsOrSchemas: pipesOrVOsOrSchemas,
                  pipes: _pipes,
              })
          );
  
          // define exec in constructor to ensure method visibility
          Object.defineProperty(this, "exec", {
              // value: (data) => pipes.get(this).exec(data),
              value: (data) => {
                  return _pipes.get(this).exec(data);
              },
              enumerable: true,
              configurable: false,
          });
      }
  
      /**
       * Creates new `pipeline` segment
       * @param pipesOrSchemas
       * @returns {Pipeline}
       */
      pipe(...pipesOrSchemas) {
          const __ = new Pipeline(...pipesOrSchemas);
          this.subscribe({
              next: (d) => {
                  __.write(d);
              }
          });
          return __;
      }
  
      /**
       * Returns arr
       * @returns {*[]}
       */
      get schema() {
          return [
              _pipes.get(this).vo.schema,
              _pipes.get(this).out.schema
          ];
      }
  
      /**
       * links pipeline segment to direct output to target pipeline
       * @param target
       * @param callbacks function[]
       * @returns {Pipeline}
       */
      link(target, ...callbacks) {
          if (!(target instanceof Pipeline)) {
              throw `item for "target" was not a Pipe`;
          }
  
          // allow for array literal in place of inline assignment
          if (Array.isArray(callbacks[0])) {
              callbacks = callbacks[0];
          }
  
          callbacks = fill(callbacks || []);
  
          // creates observer and stores it to links map for `pipeline`
          const _sub = this.subscribe({
              next: (data) => {
                  const _res = Executor.exec(callbacks, data.toJSON ? data.toJSON() : data);
                  if (_res instanceof Promise) {
                      return _res.then((_) => target.write(_));
                  }
  
                  target.write(_res);
              },
              error: (e) => {
                  console.error(e);
              },
              // handles unlink & cleanup on complete
              complete: () => this.unlink(target)
          });
  
          _pipes.get(this).links.set(target, _sub);
          return this;
      }
  
      /**
       * Unlink `pipeline` segment from target `pipeline`
       * @param target
       * @returns {Pipeline}
       */
      unlink(target) {
          if (!(target instanceof Pipeline)) {
              throw `item for "target" was not a Pipe`;
          }
  
          const _sub = _pipes.get(this).links.get(target);
  
          if (_sub) {
              _sub.unsubscribe();
              _pipes.get(this).links.delete(target);
          }
  
          return this;
      }
  
      /**
       * Returns validation errors
       * @returns {*|null}
       */
      get errors() {
          return _pipes.get(this).vo.errors || null;
      }
  
      /**
       * Returns JSON-SCHEMA for `pipeline` output
       * @returns {object}
       */
      get schemas() {
          return [..._pipes.get(this).schemas];
      }
  
      /**
       * Creates array of new `pipeline` segments that run in parallel
       * @param schemasOrPipes
       * @returns {*}
       */
      split(schemasOrPipes) {
          return schemasOrPipes.map((_) => this.pipe(_));
      }
  
      /**
       * Iterates pipeline callbacks via generator function
       * @param data
       * @returns {generator}
       */
      yield(data) {
          let _fill = _pipes.get(this).pOS.map((_) => _.exec || ((_) => _));
  
          if (!_fill.length) {
              _fill[0] = (d) => d;
          }
  
          const _f = new Function("$scope", "$cb",
              [
                  "return (function* (data) { ",
                  "try {",
                  Object.keys(_fill)
                      .map((_) => `yield data=($cb[${_}].bind($scope))(data)`)
                      .join("; "),
                  "} catch (e) { $scope.error(e); }",
                  "}).bind($scope);",
              ].join(" ")
          );
  
          return _f(this, _fill)(data);
      }
  
      /**
       * Merges multiple pipes into single output
       * @param pipeOrPipes
       * @param pipeOrSchema
       * @returns {Pipeline}
       */
      merge(pipeOrPipes, pipeOrSchema = {schemas: [DefaultVOSchema]}) {
          const _out = new Pipeline(pipeOrSchema);
          _pipes.get(this).listeners = [
              ..._pipes.get(this).listeners,
              // -- feeds output of map to listeners array
              ...(Array.isArray(pipeOrPipes) ? pipeOrPipes : [pipeOrPipes])
                  .filter((_p) => ((typeof _p.subscribe) === "function"))
                  .map((_p) => {
                      _p.subscribe((d) => {
                          // -- all pipes now write to output tx
                          _out.write(d.toJSON ? d.toJSON() : d);
                      });
                      return _p;
                  })
          ];
          // -- returns output tx for observation
          return _out;
      }
  
      /**
       * Writes data to pipeline segment
       * @param data
       * @returns {Pipeline}
       */
      write(data) {
          _pipes.get(this).vo.model = data;
          return this;
      }
  
      /**
       * Creates clone of current `pipeline` segment
       * @returns {Pipeline}
       */
      clone() {
          const $ref = _pipes.get(this);
          const _cz = class extends Pipeline {
              constructor() {
                  super();
                  _pipes.set(this, $ref);
                  _pipes.get(this).listeners = [...$ref.listeners];
              }
          };
          return new _cz();
      }
  
      /**
       * Terminates input on `pipeline` segment. This is irrevocable
       * @returns {Pipeline}
       */
      close() {
          _pipes.get(this).out.freeze();
          return this;
      }
  
      /**
       * Returns write status of `pipeline`
       * @returns {boolean}
       */
      get writable() {
          return !_pipes.get(this).out.isFrozen;
      }
  
      /**
       * Informs `Pipeline` to rate limit notifications based on time interval
       * @param rate
       * @returns {Pipeline}
       */
      throttle(rate) {
          if (rate > 0) {
              clearPipeInterval(this);
              const __ = setInterval(
                  () => {
                      if (_cache.get(this).length) {
                          const _func = _cache.get(this).splice(0, 1);
                          if ((typeof _func[0]) === "function") {
                              const _res = _func[0]();
                              _pipes.get(this).out.model = _res;
                          }
                      }
                  }, parseInt(rate, 10)
              );
  
              Object.assign(_pipes.get(this), {tO: __});
          } else if (rate === -1) {
              this.unthrottle(true);
          } else {
              this.unthrottle();
          }
  
          return this;
      }
  
      /**
       * Removes rate limiting from `Pipeline` and optionally deletes unprocessed cache items
       * @param discardCacheQueue
       */
      unthrottle(discardCacheQueue=false) {
          clearPipeInterval(this);
          if (!discardCacheQueue) {
              _cache.get(this).forEach(() => {
                  const _func = _cache.get(this).splice(0, 1);
                  if ((typeof _func[0]) === "function") {
                      _pipes.get(this).out.model = _func[0]();
                  }
              });
          } else {
              const _c = _cache.get(this);
              _c.splice(0, _c.length);
          }
      }
  
      /**
       * Returns product of Nth occurrence of `pipeline` execution
       * @param nth
       * @returns {Pipeline}
       */
      sample(nth) {
          _pipes.get(this).ivl = nth;
          return this;
      }
  
      /**
       * Subscribes to `pipeline` output notifications
       * @param handler
       * @returns {Observable}
       */
      subscribe(handler) {
          if (!(typeof handler).match(/^(function|object)$/)) {
              throw "handler required for Pipeline::subscribe";
          }
  
          return _pipes.get(this).out.subscribe(handler);
      }
  
      /**
       * Provides current state of `pipeline` output. alias for `toJSON`
       * @returns {Object|Array}
       */
      tap() {
          return this.toJSON();
      }
  
      /**
       * Convenience Method for Promise based flows.
       * Writes data to `pipeline` and wraps observer in Promise
       *
       * @param data
       * @returns {Promise<Pipeline>}
       */
      async promise(data) {
          return await new Promise((resolve, reject) => {
              this.subscribe({
                  next: (d) => {
                      resolve(d);
                  },
                  error: (e) => {
                      reject(e);
                  }
              });
              this.write(data);
          });
      }
  
      /**
       * Overrides Object's toString method
       * @override
       * @returns {String}
       */
      toString() {
          return JSON.stringify(this.toJSON());
      }
  
      /**
       * Provides current state of `pipeline` output.
       * @override
       * @returns {Object|Array}
       */
      toJSON() {
          return _pipes.get(this).out.toJSON();
      }
  }
  
  export class PipeListener {
      /**
       *
       * @param target
       */
      constructor(target, vo) {
          const _self = this;
  
          Object.defineProperties(this, {
              vo: {
                  get: () => vo,
                  enumerable: true,
                  configurable: false,
                  // writable: false,
              }
          })
  
          _pipes.set(this, target);
          this.vo.subscribe({
              next: (d) => _self.next(d),
              error: (e) => _self.error(e),
              complete: () => _self.complete(),
          });
      }
  
      /**
       *
       * @returns {Pipeline}
       */
      get target() {
          return _pipes.get(this);
      }
  
      /**
       *
       * @returns {Validator}
       */
      get vo() {
          return _pipes.get(_pipes.get(this)).vo;
      }
  
      /**
       *
       * @returns {Validator}
       */
      get out() {
          return _pipes.get(this.target).out;
      }
  
      /**
       *
       * @param e
       */
      error(e) {
          // sends error notification through out validator's observable
          _observers.get(this.out).error(e);
      }
  
      /**
       * closes `pipeline` on complete notification
       */
      complete() {
          this.target.close();
      }
  
      /**
       *
       * @param data
       * @returns {Promise<void | never>}
       */
      next(data) {
          // enforces JSON formatting if feature is present
          data = data && data.toJSON ? data.toJSON() : data;
          const _targetProps = _pipes.get(this.target);
          // tests for presence of rate-limit timeout
          if (_targetProps.tO) {
              const __ = () => {
                  return _pipes.get(this.target).exec(data);
              };
              // caches operation for later execution. Exec ordering is FIFO
              _cache.get(this.target).splice(_cache.get(this.target).length, 0, __);
              // cancels current execution
              return void 0;
          }
  
          const _target = _pipes.get(this);
  
          // tests for interval (ivl)
          if (_pipes.get(_target).ivl !== 0) {
              // tics the counter and tests if count is fulfilled
              if ((++_pipes.get(_target).ivlVal) !== _pipes.get(_target).ivl) {
                  // count is not fulfilled. stops the execution
                  return void 0;
              } else {
                  // resets the count and lets the operation proceed
                  _pipes.get(_target).ivlVal = 0;
              }
          }
  
          let _t, _type;
          const _out = (_) => {
              // else we set the model for validation
              try {
                  this.out.model = _.toJSON ? _.toJSON() : _;
              } catch (e) {
                  _observers.get(this.out).error({
                      error: e,
                      data: data,
                  });
              }
          };
  
          // capture output of callback
          try {
              _t = _pipes.get(this).exec(data); //_pipes.get(_pipes.get(this)).exec(data);
              _type = typeof _t;
          } catch (e) {
              return _observers.get(this.out).error({
                  error: e,
                  data: data,
              });
          }
  
          // tests if object and if object is writable
          if ((_t instanceof Promise) || ((_type === "function") || (_type === "object")) && _target.writable) {
              if (_t instanceof Promise) {
                  return _t.then((_) => {
                      _out(_)
                  })
                      .catch((e) => {
                          _observers.get(this.out).error({
                              error: e,
                              data: data,
                          });
                      });
              }
  
              if (_type === "function") {
                  const __ = _t();
                  if (__ instanceof Promise) {
                      return __.then((_) => {
                          _out(_)
                      }).catch((e) => {
                          _observers.get(this.out).error({
                              error: e,
                              data: data,
                          });
                      });
                  }
              }
          }
  
          _out(_t);
      }
  
      subscribe(handler) {
          _observers.get(this.out).subscribe(handler);
      }
  }
  