node-bunyan-lite/bin/bunyan

#!/usr/bin/env node
//
// bunyan -- filter and pretty-print JSON logs, like Bunyan logs.
//
// See <https://github.com/trentm/node-bunyan>.
//

var VERSION = "0.14.2";

var util = require('util');
var pathlib = require('path');
var vm = require('vm');
var http = require('http');
var fs = require('fs');
var warn = console.warn;



//---- globals and constants

// Output modes.
var OM_PAUL = 1;
var OM_JSON = 2;
var OM_INSPECT = 3;
var OM_SIMPLE = 4;
var OM_SHORT = 5;
var OM_FROM_NAME = {
  "paul": OM_PAUL,
  "json": OM_JSON,
  "inspect": OM_INSPECT,
  "simple": OM_SIMPLE,
  "short": OM_SHORT
};


// Levels
var TRACE = 10;
var DEBUG = 20;
var INFO = 30;
var WARN = 40;
var ERROR = 50;
var FATAL = 60;

var levelFromName = {
  'trace': TRACE,
  'debug': DEBUG,
  'info': INFO,
  'warn': WARN,
  'error': ERROR,
  'fatal': FATAL
};
var nameFromLevel = {};
var upperNameFromLevel = {};
var upperPaddedNameFromLevel = {};
Object.keys(levelFromName).forEach(function (name) {
  var lvl = levelFromName[name];
  nameFromLevel[lvl] = name;
  upperNameFromLevel[lvl] = name.toUpperCase();
  upperPaddedNameFromLevel[lvl] = (
    name.length === 4 ? ' ' : '') + name.toUpperCase();
});



//---- support functions

function getVersion() {
  return VERSION;
}


var format = util.format;
if (!format) {
  // If not node 0.6, then use its `util.format`:
  // <https://github.com/joyent/node/blob/master/lib/util.js#L22>:
  var inspect = util.inspect;
  var formatRegExp = /%[sdj%]/g;
  format = function format(f) {
    if (typeof f !== 'string') {
      var objects = [];
      for (var i = 0; i < arguments.length; i++) {
        objects.push(inspect(arguments[i]));
      }
      return objects.join(' ');
    }

    var i = 1;
    var args = arguments;
    var len = args.length;
    var str = String(f).replace(formatRegExp, function(x) {
      if (i >= len) return x;
      switch (x) {
        case '%s': return String(args[i++]);
        case '%d': return Number(args[i++]);
        case '%j': return JSON.stringify(args[i++]);
        case '%%': return '%';
        default:
          return x;
      }
    });
    for (var x = args[i]; i < len; x = args[++i]) {
      if (x === null || typeof x !== 'object') {
        str += ' ' + x;
      } else {
        str += ' ' + inspect(x);
      }
    }
    return str;
  };
}

function indent(s) {
  return '    ' + s.split(/\r?\n/).join('\n    ');
}

function objCopy(obj) {
  if (obj === null) {
    return null;
  } else if (Array.isArray(obj)) {
    return obj.slice();
  } else {
    var copy = {};
    Object.keys(obj).forEach(function (k) {
      copy[k] = obj[k];
    });
    return copy;
  }
}

function printHelp() {
  console.log("Usage:");
  console.log("  bunyan [OPTIONS] [FILE ...]");
  console.log("");
  console.log("Filter and pretty-print Bunyan log file content.");
  console.log("");
  console.log("General options:");
  console.log("  -h, --help    print this help info and exit");
  console.log("  --version     print version of this command and exit");
  console.log("");
  console.log("Filtering options:");
  console.log("  -l, --level LEVEL");
  console.log("                Only show messages at or above the specified level.");
  console.log("                You can specify level *names* or numeric values.");
  console.log("                (See 'Log Levels' below.)");
  console.log("  -c, --condition CONDITION");
  console.log("                Run each log message through the condition");
  console.log("                and only show those that resolve to a truish value.");
  console.log("                E.g. `-c 'pid == 123'`, `-c 'level == 50'`. You must");
  console.log("                use the numeric values for filtering by level.");
  console.log("  --strict      Suppress all but legal Bunyan JSON log lines. By default");
  console.log("                non-JSON, and non-Bunyan lines are passed through.");
  console.log("");
  console.log("Output options:");
  console.log("  --color       Colorize output. Defaults to try if output");
  console.log("                stream is a TTY.");
  console.log("  --no-color    Force no coloring (e.g. terminal doesn't support it)");
  console.log("  -o, --output MODE");
  console.log("                Specify an output mode/format. One of");
  console.log("                  paul: (the default) pretty");
  console.log("                  json: JSON output, 2-space indent");
  console.log("                  json-N: JSON output, N-space indent, e.g. 'json-4'");
  console.log("                  inspect: node.js `util.inspect` output");
  console.log("                  short: like paul, but more concise");
  console.log("  -j            shortcut for `-o json`");
  console.log("");
  console.log("Log Levels:");
  console.log("  Either numeric values or their associated strings are valid for the");
  console.log("  -l|--level argument.  However, -c|--condition scripts will see a numeric");
  console.log("  'level' value, not a string.");
  console.log("");
  Object.keys(levelFromName).forEach(function(name) {
    var n = name;
    while (n.length < 6)
      n += " ";
    console.log("    %s %d", n, levelFromName[name]);
  });
  console.log("");
  console.log("See <https://github.com/trentm/node-bunyan> for more complete docs.");
  console.log("Please report bugs to <https://github.com/trentm/node-bunyan/issues>.");
}

/*
 * If the user specifies multiple input sources, we want to print out records
 * from all sources in a single, chronologically ordered stream.  To do this
 * efficiently, we first assume that all records within each source are ordered
 * already, so we need only keep track of the next record in each source and
 * the time of the last record emitted.  To avoid excess memory usage, we
 * pause() streams that are ahead of others.
 *
 * "streams" is an object indexed by source name (file name) which specifies:
 *
 *    stream        Actual stream object, so that we can pause and resume it.
 *
 *    records       Array of log records we've read, but not yet emitted.  Each
 *                  record includes "line" (the raw line), "rec" (the JSON
 *                  record), and "time" (the parsed time value).
 *
 *    done          Whether the stream has any more records to emit.
 */
var streams = {};

function gotRecord(file, line, rec, opts, stylize)
{
  var time = new Date(rec.time);

  streams[file]['records'].push({ line: line, rec: rec, time: time });
  emitNextRecord(opts, stylize);
}

function filterRecord(rec, opts)
{
  if (opts.level && rec.level < opts.level) {
    return false;
  }

  if (opts.conditions) {
    for (var i = 0; i < opts.conditions.length; i++) {
      var pass = opts.conditions[i].runInNewContext(rec);
      if (!pass)
        return false;
    }
  }

  return true;
}

function emitNextRecord(opts, stylize)
{
  var ofile, ready, minfile, rec;

  for (;;) {
    /*
     * Take a first pass through the input streams to see if we have a record
     * from all of them.  If not, we'll pause any streams for which we do
     * already have a record (to avoid consuming excess memory) and then wait
     * until we have records from the others before emitting the next record.
     *
     * As part of the same pass, we look for the earliest record we have not yet
     * emitted.
     */
    minfile = undefined;
    ready = true;
    for (ofile in streams) {

      if (streams[ofile].stream === null ||
        (!streams[ofile].done && streams[ofile].records.length === 0)) {
        ready = false;
        break;
      }

      if (streams[ofile].records.length > 0 && (minfile === undefined ||
          streams[minfile].records[0].time > streams[ofile].records[0].time)) {
        minfile = ofile;
      }
    }

    if (!ready || minfile === undefined) {
      for (ofile in streams) {
        if (!streams[ofile].stream || streams[ofile].done)
          continue;

        if (streams[ofile].records.length > 0) {
          if (!streams[ofile].paused) {
            streams[ofile].paused = true;
            streams[ofile].stream.pause();
          }
        } else if (streams[ofile].paused) {
          streams[ofile].paused = false;
          streams[ofile].stream.resume();
        }
      }

      return;
    }

    /*
     * Emit the next record for "minfile", and invoke ourselves again to make
     * sure we emit as many records as we can right now.
     */
    rec = streams[minfile].records.shift();
    emitRecord(rec.rec, rec.line, opts, stylize);
  }
}

/**
 * Parse the command-line options and arguments into an object.
 *
 *    {
 *      'args': [...]       // arguments
 *      'help': true,       // true if '-h' option given
 *       // etc.
 *    }
 *
 * @return {Object} The parsed options. `.args` is the argument list.
 * @throws {Error} If there is an error parsing argv.
 */
function parseArgv(argv) {
  var parsed = {
    args: [],
    help: false,
    color: process.stdout.isTTY,
    outputMode: OM_PAUL,
    jsonIndent: 2,
    level: null,
    conditions: null,
    strict: false
  };

  // Turn '-iH' into '-i -H', except for argument-accepting options.
  var args = argv.slice(2);  // drop ['node', 'scriptname']
  var newArgs = [];
  var optTakesArg = {'d': true, 'o': true, 'c': true, 'l': true};
  for (var i = 0; i < args.length; i++) {
    if (args[i].charAt(0) === "-" && args[i].charAt(1) !== '-' && args[i].length > 2) {
      var splitOpts = args[i].slice(1).split("");
      for (var j = 0; j < splitOpts.length; j++) {
        newArgs.push('-' + splitOpts[j]);
        if (optTakesArg[splitOpts[j]]) {
          var optArg = splitOpts.slice(j+1).join("");
          if (optArg.length) {
            newArgs.push(optArg);
          }
          break;
        }
      }
    } else {
      newArgs.push(args[i]);
    }
  }
  args = newArgs;

  var condDefines = [];
  Object.keys(upperNameFromLevel).forEach(function (lvl) {
    condDefines.push(
      format("Object.prototype.%s = %s;", upperNameFromLevel[lvl], lvl));
  });
  condDefines = condDefines.join('\n') + '\n';

  var endOfOptions = false;
  while (args.length > 0) {
    var arg = args.shift();
    switch(arg) {
      case "--":
        endOfOptions = true;
        break;
      case "-h": // display help and exit
      case "--help":
        parsed.help = true;
        break;
      case "--version":
        parsed.version = true;
        break;
      case "--strict":
        parsed.strict = true;
        break;
      case "--color":
        parsed.color = true;
        break;
      case "--no-color":
        parsed.color = false;
        break;
      case "-o":
      case "--output":
        var name = args.shift();
        var idx = name.lastIndexOf('-');
        if (idx !== -1) {
          var indent = Number(name.slice(idx+1));
          if (! isNaN(indent)) {
            parsed.jsonIndent = indent;
            name = name.slice(0, idx);
          }
        }
        parsed.outputMode = OM_FROM_NAME[name];
        if (parsed.outputMode === undefined) {
          throw new Error("unknown output mode: '"+name+"'");
        }
        break;
      case "-j": // output with JSON.stringify
        parsed.outputMode = OM_JSON;
        break;
      case "-l":
      case "--level":
        var levelArg = args.shift();
        var level = +(levelArg);
        if (isNaN(level)) {
          level = +levelFromName[levelArg.toLowerCase()];
        }
        if (isNaN(level)) {
          throw new Error("unknown level value: '"+levelArg+"'");
        }
        parsed.level = level;
        break;
      case "-c":
      case "--condition":
        var condition = args.shift();
        parsed.conditions = parsed.conditions || [];
        var scriptName = 'bunyan-condition-'+parsed.conditions.length;
        var script = vm.createScript(condDefines + condition, scriptName);
        parsed.conditions.push(script);
        break;
      default: // arguments
        if (!endOfOptions && arg.length > 0 && arg[0] === '-') {
          throw new Error("unknown option '"+arg+"'");
        }
        parsed.args.push(arg);
        break;
    }
  }
  //TODO: '--' handling and error on a first arg that looks like an option.

  return parsed;
}


function isInteger(s) {
  return (s.search(/^-?[0-9]+$/) == 0);
}


// http://en.wikipedia.org/wiki/ANSI_escape_code#graphics
// Suggested colors (some are unreadable in common cases):
// - Good: cyan, yellow (limited use), grey, bold, green, magenta, red
// - Bad: blue (not visible on cmd.exe)
var colors = {
  'bold' : [1, 22],
  'italic' : [3, 23],
  'underline' : [4, 24],
  'inverse' : [7, 27],
  'white' : [37, 39],
  'grey' : [90, 39],
  'black' : [30, 39],
  'blue' : [34, 39],
  'cyan' : [36, 39],
  'green' : [32, 39],
  'magenta' : [35, 39],
  'red' : [31, 39],
  'yellow' : [33, 39]
};

function stylizeWithColor(str, color) {
  if (!str)
    return '';
  var codes = colors[color];
  if (codes) {
    return '\033[' + codes[0] + 'm' + str +
           '\033[' + codes[1] + 'm';
  } else {
    return str;
  }
}

function stylizeWithoutColor(str, color) {
  return str;
}


/**
 * Is this a valid Bunyan log record.
 */
function isValidRecord(rec) {
  if (rec.v == null ||
      rec.level == null ||
      rec.name == null ||
      rec.hostname == null ||
      rec.pid == null ||
      rec.time == null ||
      rec.msg == null) {
    // Not valid Bunyan log.
    return false;
  } else {
    return true;
  }
}


/**
 * Parses the given log line and either emits it right away (for invalid
 * records) or enqueues it for emitting later when it's the next line to show.
 */
function handleLogLine(file, line, opts, stylize) {
  // Emit non-JSON lines immediately.
  var rec;
  if (!line) {
    if (!opts.strict) emit(line + '\n');
    return;
  } else if (line[0] !== '{') {
    if (!opts.strict) emit(line + '\n');  // not JSON
    return;
  } else {
    try {
      rec = JSON.parse(line);
    } catch(e) {
      if (!opts.strict) emit(line + '\n');
      return;
    }
  }

  if (!isValidRecord(rec)) {
    if (!opts.strict) emitRecord(rec, line, opts, stylize);
    return;
  }

  if (!filterRecord(rec, opts))
    return;

  if (file === null)
    return emitRecord(rec, line, opts, stylize);

  return gotRecord(file, line, rec, opts, stylize);
}

/**
 * Print out a single result, considering input options.
 */
function emitRecord(rec, line, opts, stylize) {
  var short = false;

  switch (opts.outputMode) {
  case OM_SHORT:
    short = true;
    /* jsl:fall-thru */

  case OM_PAUL:
    //    [time] LEVEL: name[/component]/pid on hostname (src): msg* (extras...)
    //        msg*
    //        --
    //        long and multi-line extras
    //        ...
    // If 'msg' is single-line, then it goes in the top line.
    // If 'req', show the request.
    // If 'res', show the response.
    // If 'err' and 'err.stack' then show that.
    if (!isValidRecord(rec)) {
      return emit(line + '\n');
    }

    delete rec.v;

    /*
     * We assume the Date is formatted according to ISO8601, in which case we
     * can safely chop off the date information.
     */
    if (short && rec.time[10] == 'T') {
      var time = rec.time.substr(11);
      time = stylize(time, 'XXX');
    } else {
      var time = stylize('[' + rec.time + ']', 'XXX');
    }

    delete rec.time;

    var nameStr = rec.name;
    delete rec.name;

    if (rec.component) {
      nameStr += '/' + rec.component;
    }
    delete rec.component;

    if (!short)
      nameStr += '/' + rec.pid;
    delete rec.pid;

    var level = (upperPaddedNameFromLevel[rec.level] || "LVL" + rec.level);
    if (opts.color) {
      var colorFromLevel = {
        10: 'grey',     // TRACE
        20: 'grey',     // DEBUG
        30: 'cyan',     // INFO
        40: 'magenta',  // WARN
        50: 'red',      // ERROR
        60: 'inverse',  // FATAL
      };
      level = stylize(level, colorFromLevel[rec.level]);
    }
    delete rec.level;

    var src = "";
    if (rec.src && rec.src.file) {
      var s = rec.src;
      if (s.func) {
        src = format(" (%s:%d in %s)", s.file, s.line, s.func);
      } else {
        src = format(" (%s:%d)", s.file, s.line);
      }
      src = stylize(src, 'green');
    }
    delete rec.src;

    var hostname = rec.hostname;
    delete rec.hostname;

    var extras = [];
    var details = [];

    if (rec.req_id) {
      extras.push("req_id=" + rec.req_id);
    }
    delete rec.req_id;

    if (rec.latency) {
      extras.push(rec.latency + "ms");
    }
    delete rec.latency;

    var onelineMsg;
    if (rec.msg.indexOf('\n') !== -1) {
      onelineMsg = '';
      details.push(indent(stylize(rec.msg, 'cyan')));
    } else {
      onelineMsg = ' ' + stylize(rec.msg, 'cyan');
    }
    delete rec.msg;

    if (rec.req && typeof(rec.req) === 'object') {
      var req = rec.req;
      delete rec.req;
      var headers = req.headers;
      var s = format("%s %s HTTP/%s%s", req.method,
        req.url,
        req.httpVersion || "1.1",
        (headers
          ? '\n' + Object.keys(headers).map(
              function (h) { return h + ': ' + headers[h]; }).join('\n')
          : '')
      );
      delete req.url;
      delete req.method;
      delete req.httpVersion;
      delete req.headers;
      if (req.body) {
        s += '\n\n' + (typeof(req.body) === 'object'
          ? JSON.stringify(req.body, null, 2) : req.body);
        delete req.body;
      }
      if (req.trailers && Object.keys(req.trailers) > 0) {
        s += '\n' + Object.keys(req.trailers).map(
            function (t) { return t + ': ' + req.trailers[t]; }).join('\n');
      }
      delete req.trailers;
      details.push(indent(s));
      // E.g. for extra 'foo' field on 'req', add 'req.foo' at top-level.
      // This *does* have the potential to stomp on a literal 'req.foo' key.
      Object.keys(req).forEach(function (k) {
        rec["req." + k] = req[k];
      })
    }

    if (rec.client_req && typeof(rec.client_req) === 'object') {
      var client_req = rec.client_req;
      delete rec.client_req;
      var headers = client_req.headers;
      var hostHeaderLine = '';
      var s = '';
      if (client_req.address) {
        hostHeaderLine = 'Host: ' + client_req.address;
        if (client_req.port)
          hostHeaderLine += ':' + client_req.port;
        hostHeaderLine += '\n';
      }
      delete client_req.headers;
      delete client_req.address;
      delete client_req.port;
      s += format("%s %s HTTP/%s\n%s%s", client_req.method,
        client_req.url,
        client_req.httpVersion || "1.1",
        hostHeaderLine,
        Object.keys(headers).map(
          function (h) { return h + ': ' + headers[h]; }).join('\n'));
      delete client_req.method;
      delete client_req.url;
      delete client_req.httpVersion;
      if (client_req.body) {
        s += '\n\n' + (typeof(client_req.body) === 'object'
          ? JSON.stringify(client_req.body, null, 2) : client_req.body);
        delete client_req.body;
      }
      // E.g. for extra 'foo' field on 'client_req', add 'client_req.foo' at
      // top-level. This *does* have the potential to stomp on a literal
      // 'client_req.foo' key.
      Object.keys(client_req).forEach(function (k) {
        rec["client_req." + k] = client_req[k];
      })
      details.push(indent(s));
    }

    if (rec.res && typeof(rec.res) === 'object') {
      var res = rec.res;
      delete rec.res;
      var s = '';
      if (res.header) {
        s += res.header.trimRight();
      } else if (res.headers) {
        if (res.statusCode) {
          s += format("HTTP/1.1 %s %s\n", res.statusCode,
            http.STATUS_CODES[res.statusCode]);
        }
        var headers = res.headers;
        s += Object.keys(headers).map(
          function (h) { return h + ': ' + headers[h]; }).join('\n');
      }
      delete res.header;
      delete res.headers;
      delete res.statusCode;
      if (res.body) {
        s += '\n\n' + (typeof(res.body) === 'object'
          ? JSON.stringify(res.body, null, 2) : res.body);
        delete res.body;
      }
      if (res.trailer) {
        s += '\n' + res.trailer;
      }
      delete res.trailer;
      if (s) {
        details.push(indent(s));
      }
      // E.g. for extra 'foo' field on 'res', add 'res.foo' at top-level.
      // This *does* have the potential to stomp on a literal 'res.foo' key.
      Object.keys(res).forEach(function (k) {
        rec["res." + k] = res[k];
      });
    }

    if (rec.err && rec.err.stack) {
      details.push(indent(rec.err.stack));
      delete rec.err;
    }

    var leftover = Object.keys(rec);
    for (var i = 0; i < leftover.length; i++) {
      var key = leftover[i];
      var value = rec[key];
      var stringified = false;
      if (typeof(value) !== 'string') {
        value = JSON.stringify(value, null, 2);
        stringified = true;
      }
      if (value.indexOf('\n') !== -1 || value.length > 50) {
        details.push(indent(key + ': ' + value));
      } else if (!stringified && (value.indexOf(' ') != -1 ||
                                  value.length === 0)) {
        extras.push(key + '=' + JSON.stringify(value));
      } else {
        extras.push(key + '=' + value);
      }
    }

    extras = stylize(
      (extras.length ? ' (' + extras.join(', ') + ')' : ''), 'grey');
    details = stylize(
      (details.length ? details.join('\n    --\n') + '\n' : ''), 'grey');
    if (!short)
      emit(format("%s %s: %s on %s%s:%s%s\n%s",
        time,
        level,
        nameStr,
        hostname || "<no-hostname>",
        src,
        onelineMsg,
        extras,
        details));
    else
      emit(format("%s %s %s:%s%s\n%s",
        time,
        level,
        nameStr,
        onelineMsg,
        extras,
        details));
    break;

  case OM_INSPECT:
    emit(util.inspect(rec, false, Infinity, true) + '\n');
    break;

  case OM_JSON:
    emit(JSON.stringify(rec, null, opts.jsonIndent) + '\n');
    break;

  case OM_SIMPLE:
    // <http://logging.apache.org/log4j/1.2/apidocs/org/apache/log4j/SimpleLayout.html>
    if (!isValidRecord(rec)) {
      return emit(line + '\n');
    }
    emit(format("%s - %s\n", upperNameFromLevel[rec.level] || "LVL" + rec.level,
      rec.msg));
    break;
  default:
    throw new Error("unknown output mode: "+opts.outputMode);
  }
}


var stdoutFlushed = true;
function emit(s) {
  // TODO:PERF If this is try/catch is too slow (too granular): move up to
  //    mainline and be sure to only catch this particular error.
  try {
    stdoutFlushed = process.stdout.write(s);
  } catch (e) {
    // Handle any exceptions in stdout writing in the "error" event above.
  }
}

process.stdout.on("error", function (err) {
  if (err.code === "EPIPE") {
    // Pass. See <https://github.com/trentm/json/issues/9>.
  } else {
    warn(err);
    drainStdoutAndExit(1);
  }
});


/**
 * A hacked up version of "process.exit" that will first drain stdout
 * before exiting. *WARNING: This doesn't stop event processing.* IOW,
 * callers have to be careful that code following this call isn't
 * accidentally executed.
 *
 * In node v0.6 "process.stdout and process.stderr are blocking when they
 * refer to regular files or TTY file descriptors." However, this hack might
 * still be necessary in a shell pipeline.
 */
function drainStdoutAndExit(code) {
  process.stdout.on('drain', function () {
    process.exit(code);
  });
  if (stdoutFlushed) {
    process.exit(code);
  }
}


/**
 * Process all input from stdin.
 *
 * @params opts {Object} Bunyan options object.
 * @param stylize {Function} Output stylize function to use.
 * @param callback {Function} `function ()`
 */
function processStdin(opts, stylize, callback) {
  var leftover = "";  // Left-over partial line from last chunk.
  var stdin = process.stdin;
  stdin.resume();
  stdin.setEncoding('utf8');
  stdin.on('data', function (chunk) {
    var lines = chunk.split(/\r\n|\n/);
    var length = lines.length;
    if (length === 1) {
      leftover += lines[0];
      return;
    }

    if (length > 1) {
      handleLogLine(null, leftover + lines[0], opts, stylize);
    }
    leftover = lines.pop();
    length -= 1;
    for (var i=1; i < length; i++) {
      handleLogLine(null, lines[i], opts, stylize);
    }
  });
  stdin.on('end', function () {
    if (leftover) {
      handleLogLine(null, leftover, opts, stylize);
      leftover = '';
    }
    callback();
  });
}


/**
 * Process all input from the given log file.
 *
 * @param file {String} Log file path to process.
 * @params opts {Object} Bunyan options object.
 * @param stylize {Function} Output stylize function to use.
 * @param callback {Function} `function ()`
 */
function processFile(file, opts, stylize, callback) {
  var stream = fs.createReadStream(file);
  if (/\.gz$/.test(file)) {
    stream = stream.pipe(require('zlib').createGunzip());
  }
  // Manually decode streams - lazy load here as per node/lib/fs.js
  var decoder = new (require('string_decoder').StringDecoder)('utf8');

  streams[file].stream = stream;

  stream.on('error', function (err) {
    streams[file].done = true;
    callback(err);
  });

  var leftover = '';  // Left-over partial line from last chunk.
  stream.on('data', function (data) {
    var chunk = decoder.write(data);
    if (!chunk.length) {
      return;
    }
    var lines = chunk.split(/\r\n|\n/);
    var length = lines.length;
    if (length === 1) {
      leftover += lines[0];
      return;
    }

    if (length > 1) {
      handleLogLine(file, leftover + lines[0], opts, stylize);
    }
    leftover = lines.pop();
    length -= 1;
    for (var i=1; i < length; i++) {
      handleLogLine(file, lines[i], opts, stylize);
    }
  });

  stream.on('end', function () {
    streams[file].done = true;
    if (leftover) {
      handleLogLine(file, leftover, opts, stylize);
      leftover = '';
    } else {
      emitNextRecord(opts, stylize);
    }
    callback();
  });
}


/**
 * From node async module.
 */
function asyncForEach(arr, iterator, callback) {
    callback = callback || function () {};
    if (!arr.length) {
        return callback();
    }
    var completed = 0;
    arr.forEach(function (x) {
        iterator(x, function (err) {
            if (err) {
                callback(err);
                callback = function () {};
            }
            else {
                completed += 1;
                if (completed === arr.length) {
                    callback();
                }
            }
        });
    });
};



//---- mainline

process.on('uncaughtException', function (err) {
  function indent(s) {
    var lines = s.split(/\r?\n/);
    for (var i = 0; i < lines.length; i++) {
      lines[i] = '*     ' + lines[i];
    }
    return lines.join('\n');
  }

  console.error('* * *');
  console.error('* The Bunyan CLI crashed! Please report this issue:');
  console.error('*');
  console.error('*    https://github.com/trentm/node-bunyan/issues/new');
  console.error('*');
  console.error('* and include the following Bunyan details the log file');
  console.error('* (or section of log file) on which the Bunyan CLI crashed.');
  console.error('* * *');
  console.error('* node version:', process.version);
  console.error('* bunyan version:', getVersion());
  console.error('* argv: %j', process.argv);
  console.error('* stack:');
  console.error(indent(err.stack));
  console.error('* * *');
  process.exit(1);
});

function main(argv) {
  var opts;
  try {
    opts = parseArgv(argv);
  } catch (e) {
    warn("bunyan: error: %s", e.message);
    return drainStdoutAndExit(1);
  }
  if (opts.help) {
    printHelp();
    return;
  }
  if (opts.version) {
    console.log("bunyan " + getVersion());
    return;
  }
  var stylize = (opts.color ? stylizeWithColor : stylizeWithoutColor);

  var retval = 0;
  if (opts.args.length > 0) {
    var files = opts.args;
    files.forEach(function (file) {
      streams[file] = { stream: null, records: [], done: false }
    });
    asyncForEach(files,
      function (file, next) {
        processFile(file, opts, stylize, function (err) {
          if (err) {
            warn('bunyan: %s', err.message);
            retval += 1;
          }
          next();
        });
      },
      function (err) {
        if (err) {
          warn("bunyan: unexpected error: %s", err.stack || err);
          return drainStdoutAndExit(1);
        }
        process.exit(retval);
      }
    );
  } else {
    processStdin(opts, stylize, function () {
      process.exit(retval);
    });
  }
}

if (require.main === module) {
  // HACK guard for <https://github.com/trentm/json/issues/24>.
  // We override the `process.stdout.end` guard that core node.js puts in
  // place. The real fix is that `.end()` shouldn't be called on stdout
  // in node core. Node v0.6.9 fixes that. Only guard for v0.6.0..v0.6.8.
  var nodeVer = process.versions.node.split('.').map(Number);
  if ([0,6,0] <= nodeVer && nodeVer <= [0,6,8]) {
    var stdout = process.stdout;
    stdout.end = stdout.destroy = stdout.destroySoon = function() {
      /* pass */
    };
  }

  main(process.argv);
}