ndjson-top

#!/usr/bin/env node

var 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]")
    .description("Select the top values from a newline-delimited JSON stream.")
    .option("-n <value>", "the maximum number of output values", 1)
    .option("-r, --require <name=module>", "require a module", requires, {a: undefined, b: undefined})
    .parse(process.argv);

if (commander.args.length > 1) {
  commander.outputHelp();
  process.exit(1);
}

if (commander.args.length < 1) {
  commander.args[0] = "a < b ? -1 : a > b ? 1 : a >= b ? 0 : NaN";
}

var i = -1,
    sandbox = commander.require,
    heap = new Heap(commander.n, function(a, b) { return sandbox.a = a, sandbox.b = b, compare.runInContext(context); }),
    compare = expression(commander.args[0]),
    context = new vm.createContext(sandbox);

readline.createInterface({
  input: process.stdin,
  output: null
}).on("line", function(line) {
  ++i;
  try {
    heap.push(JSON.parse(line));
  } catch (error) {
    console.error("stdin:" + (i + 1));
    console.error(line);
    console.error("^");
    console.error("SyntaxError: " + error.message);
    process.exit(1);
  }
}).on("close", function() {
  var result
  while (!(result = heap.pop()).done) {
    output(result.value);
  }
});

function Heap(k, compare) {
  this._array = [];
  this._n = 0;
  this._k = k;
  this._compare = compare;
}

Heap.prototype.push = function(object) {
  heap_up(this, this._array[this._n] = object, this._n++);
  if (this._array.length > this._k) this.pop();
};

Heap.prototype.pop = function(object) {
  if (this._n <= 0) return {done: true, value: undefined};
  var removed = this._array[0], object;
  if (--this._n > 0) object = this._array[this._n], heap_down(this, this._array[0] = object, 0);
  return {done: false, value: removed};
};

function heap_up(heap, object, i) {
  while (i > 0) {
    var j = ((i + 1) >> 1) - 1, parent = heap._array[j];
    if (heap._compare(object, parent) >= 0) break;
    heap._array[i] = parent;
    heap._array[i = j] = object;
  }
}

function heap_down(heap, object, i) {
  while (true) {
    var r = (i + 1) << 1, l = r - 1, j = i, child = heap._array[j];
    if (l < heap._n && heap._compare(heap._array[l], child) < 0) child = heap._array[j = l];
    if (r < heap._n && heap._compare(heap._array[r], child) < 0) child = heap._array[j = r];
    if (j === i) break;
    heap._array[i] = child;
    heap._array[i = j] = object;
  }
}