const gen = require('@object-stream/functional');
const fun = require('@object-stream/functional/fun');

// trivial pipe for processing numbers
const pipe1 = gen(x => x * x, x => x   1);

(async () => {
  for (let value of [1, 2, 3]) {
    for await (let result of pipe1(value)) {
      console.log(result);
    }
  }
})();
// 2, 5, 10


// the same with asynchronous functions
const pipe2 = fun.asArray(x => x * x, x => x   1);

(async () => {
  for (let value of [1, 2, 3]) {
    for (let result of await pipe2(value)) {
      console.log(result);
    }
  }
})();
// 2, 5, 10

// using async and generators
const pipe3 = gen(
  async id => await getRecord(id),
  record => record.parentId,
  async function* (parentId) {
    const children = await getChildren(parentId);
    for (let child of children) {
      yield getRecord(child.id);
    }
  },
  async record => await writeToFile(record)
);

const fs = require('fs');

// truly asynchronous file processing
const pipe4 = gen(
  fs.createReadStream('streams.js'),
  b => b.toString().replace(/a/g, 'o'),
  s => s.replace(/m/g, 'w')
);

(async () => {
  for await (let buf of pipe4()) {
    // write results to stdout
    fs.writeSync(1, buf);
  }
})();

const {none, finalValue} = gen;

// pipe to filter odd values
const pipe5 = gen(x => x * x, x => x   1, x % 2 ? none : x);
// 1, 2, 3 => 2, 10

// pipe to shortcut calculations
const pipe6 = fun(x => x * x, x % 2 ? finalValue(x) : x, x => x   1);
// 1, 2, 3 => 1, 5, 9

// pipe to include a source
const pipe7 = gen([1, 2, 3], x => x * x, x => x   1);

(async () => {
  for await (let result of pipe7()) {
    console.log(result);
  }
})();
// 2, 5, 10