// Library functions:
function pipe(arg, ...fns) {
  return fns.reduce((v, fn) => fn(v), arg);
}

function fromCallback(emitter) {
  let values;
  let resolve;
  const init = r => [values, resolve] = [[], r];  
  let valuesAvailable = new Promise(init);
  emitter(value => {
    values.push(value);
    resolve(values);
  });
  return {
    [Symbol.asyncIterator]: async function*() {
      for (;;) {
        const vs = await valuesAvailable;
        valuesAvailable = new Promise(init);
        yield* vs;
      }
    }
  };
}

function fromEvent(element, event, options) {
  if (typeof element === 'string') {
    element = document.querySelector(element);
  }
  return element ? fromCallback(listener => {
    element.addEventListener(event, listener, options);
  }) : [];
}
  
function dup(count = 2) {
  return function (src) {
    const dups = [];

    async function pull() {
      if (Symbol.asyncIterator in src) {
        const iter = src[Symbol.asyncIterator]();
        for (;;) {
          const promise = iter.next();
          dups.forEach(d => d.promises.push(promise));
          const r = await promise;
          if (r.done) {
            return;
          }
        }
      }
      if (Symbol.iterator in src) {
        for (const value of src) {
          const promise = Promise.resolve({value});
          dups.forEach(d => d.promises.push(promise));
        }
      }
      const promise = Promise.resolve({done: true});
      dups.forEach(d => d.promises.push(promise));   
    }

    for (let i = 0; i < count; ++i) {
      const dup = {
        promises: [],
        [Symbol.asyncIterator]: () => ({
          next: () => dup.promises.shift()
        })
      };
      dups.push(dup);
    }

    pull();
    return dups;
  };
}

function reduce(op, acc) {
  return async function* (src) {
    let index = 0;
    for await (const value of src) {
      acc = op(acc, value, index++);
      yield acc;
    }
  };
}

function flat(depth = 1) {
  return async function* (src) {
    for await (const value of src) {
      if (depth > 0 && (value[Symbol.asyncIterator] || value[Symbol.iterator]))  {
        yield* flat(depth - 1)(value);
      } else {
        yield value;      
      }
    }
  };
}
    
function forEach(op) {
  return async function forEach(src) {
    let index = 0;
    for await (const value of src) {
      if (op(value, index++) === false) {
        break;
      }
    }
  };
}

function filter(op) {
  return async function* (src) {
    let index = 0;
    for await (const value of src) {
      if (op(value, index++)) {
        yield value;
      }
    }
  };
}

function map(op) {
  return async function* map(src) {
    let index = 0;
    for await (const value of src) {
      yield op(value, index++);
    }
  };
}

async function head(src) {
  for await (const value of src) {
    return value;
  }  
}

function find(op) {
  return async function (src) {
    return pipe(src, filter(op), head);
  };
}

// Test the library
async function demo() {
  const [ticker1, ticker2] = pipe(fromCallback(emitTicks), flat(), dup());
  pipe(ticker1, filter((v, i) => i > 12), forEach(endTicks));
  pipe(ticker2, forEach(v => console.log('Ticks', v)));
  
  pipe(fromEvent('button', 'click'), reduce(acc => acc + 1, 0), forEach(v => console.log('Click', v)));
}
demo();

// Helper functions for testing
function emitValues(cb) {
  const vs = [1, 2, 42, 3];
  vs.forEach(cb);
}

var intervalId;
function emitTicks(cb) {
  let i = 0;
  intervalId = setInterval(() => {
    cb([i++, i++, i++]); // Date.now());
  }, 1000);
}
function endTicks() {
  clearInterval(intervalId);
  return false;
}