ndjson-join

#!/usr/bin/env node

var fs = require("fs"),
    readline = require("readline"),
    vm = require("vm"),
    commander = require("commander"),
    requires = require("./requires"),
    expression = require("./expression"),
    output = require("./output");

commander
    .version(require("./package.json").version)
    .usage("[options] [expression₀ [expression₁]] <file₀> <file₁>")
    .description("Join two newline-delimited JSON streams into a stream of pairs.")
    .option("-r, --require <name=module>", "require a module", requires, {d: undefined, i: -1})
    .option("--left", "perform a left join", false)
    .option("--right", "perform a right join", false)
    .option("--outer", "perform an outer join", false)
    .parse(process.argv);

if (commander.args.length < 2 || commander.args.length > 4) {
  commander.outputHelp();
  process.exit(1);
}

if (commander.args.length < 4) {
  if (commander.args.length < 3) commander.args.splice(0, 0, "i");
  commander.args.splice(1, 0, commander.args[0]);
}

if ([commander.left, commander.right, commander.outer].filter(Boolean).length > 1) {
  console.error();
  console.error("  error: only one of --left, --right, --outer allowed")
  console.error();
  process.exit(1);
}

var join = commander.left ? joinLeft
    : commander.right ? joinRight
    : commander.outer ? joinOuter
    : joinInner;

var i0 = -1,
    i1 = -1,
    ii = 0,
    sandbox = commander.require,
    map = new Map,
    key0 = expression(commander.args[0]),
    key1 = expression(commander.args[1]),
    context = new vm.createContext(sandbox);

readline.createInterface({
  input: commander.args[2] === "-" ? process.stdin : fs.createReadStream(commander.args[2]),
  output: null
}).on("line", function(line) {
  sandbox.i = ++i0;
  try {
    sandbox.d = JSON.parse(line);
  } catch (error) {
    console.error("stdin:" + (i0 + 1));
    console.error(line);
    console.error("^");
    console.error("SyntaxError: " + error.message);
    process.exit(1);
  }
  var k = key0.runInContext(context);
  if (map.has(k)) map.get(k)[0].push(sandbox.d);
  else map.set(k, [[sandbox.d], []]);
}).on("close", function() {
  if ((ii |= 1) === 3) join();
});

readline.createInterface({
  input: commander.args[3] === "-" ? process.stdin : fs.createReadStream(commander.args[3]),
  output: null
}).on("line", function(line) {
  sandbox.i = ++i1;
  try {
    sandbox.d = JSON.parse(line);
  } catch (error) {
    console.error("stdin:" + (i1 + 1));
    console.error(line);
    console.error("^");
    console.error("SyntaxError: " + error.message);
    process.exit(1);
  }
  var k = key1.runInContext(context);
  if (map.has(k)) map.get(k)[1].push(sandbox.d);
  else map.set(k, [[], [sandbox.d]]);
}).on("close", function() {
  if ((ii |= 2) === 3) join();
});

function joinInner() {
  map.forEach(function(value, key) {
    value[0].forEach(function(v0) {
      value[1].forEach(function(v1) {
        output([v0, v1]);
      });
    });
  });
}

function joinLeft() {
  map.forEach(joinLeftValue);
}

function joinRight() {
  map.forEach(joinRightValue);
}

function joinOuter() {
  map.forEach(function(value, key) {
    if (value[0].length) joinLeftValue(value, key);
    else if (value[1].length) joinRightValue(value, key);
  });
}

function joinLeftValue(value, key) {
  value[0].forEach(function(v0) {
    if (value[1].length) {
      value[1].forEach(function(v1) {
        output([v0, v1]);
      });
    } else {
      output([v0, null]);
    }
  });
}

function joinRightValue(value, key) {
  value[1].forEach(function(v1) {
    if (value[0].length) {
      value[0].forEach(function(v0) {
        output([v0, v1]);
      });
    } else {
      output([null, v1]);
    }
  });
}