Add --stream option to CLI. Refs #26

Automatically switch to streaming mode when the input file is
over 150kB or manually enable with --stream or -x.

Not made the default mode since it is actually slower for files
that can be parsed using the default method without running out
of available memory. The hard limit for that is around 200kB on
my machine.

Refs #27

Author: nwronski

README.md

@@ -75,12 +75,12 @@ var readStream = require('fs').createReadStream('./large-input-file.sql');
 readStream.pipe(parserTransform);
 parserTransform.pipe(process.stdout);
 
-parser.on('error', function (err) {
+parserTransform.on('error', function (err) {
   console.log(err);
   process.exit(1);
 });
 
-parser.on('finish', function () {
+parserTransform.on('finish', function () {
   process.exit(0);
 });
 ```

bin/sqlite-parser

@@ -4,4 +4,4 @@
  * @copyright 2015-2016 Code School (http://codeschool.com)
  * @author Nick Wronski <[email protected]>
  */
-'use strict';var _index=require('../lib/index');var _index2=_interopRequireDefault(_index);var _fs=require('fs');var _path=require('path');function _interopRequireDefault(obj){return obj&&obj.__esModule?obj:{default:obj};}var args=resolveArgs(process.argv.slice(2));if(args['version']){console.log('sqlite-parser v1.0.0-beta2');process.exit(0);}if(args['help']||args._.length===0){console.log('Usage: sqlite-parser infile [--output outfile]');process.exit(0);}var input=(0,_path.normalize)(args._[0]);var output=args['o']||args['output'];if(output){output=(0,_path.normalize)(output);}(0,_fs.stat)(input,function startCallback(err){if(err){return error(err);}(0,_fs.readFile)(input,'utf8',readCallback);});function resolveArgs(argv){var args={_:[]};var last=null;var isNewArg=function isNewArg(arg){return!arg||arg.indexOf('-')===0;};var aliases={o:'output',v:'version',h:'help'};for(var i=0;i<argv.length;i+=1){var arg=argv[i];if(isNewArg(arg)){var cur=arg.indexOf('--')!==-1?arg.slice(2):aliases[arg.slice(1)];var peek=argv.length-1!==i?argv[i+1]:null;var peekNew=isNewArg(peek);args[cur]=peekNew?true:peek;if(!peekNew){i+=1;}}else{args._.push(arg);}}return args;}function error(err){console.error(err.message);process.exit(1);}function writeOut(result,outPath){var outDir=(0,_path.dirname)(outPath);function writeCallback(err){if(err){return error(err);}process.exit(0);}function mkdirCallback(err){if(err){return error(err);}(0,_fs.writeFile)(outPath,result,writeCallback);}function statCallback(err){if(err){return(0,_fs.mkdir)(outDir,mkdirCallback);}mkdirCallback();}(0,_fs.stat)(outDir,statCallback);}function parserCallback(err,ast){if(err){return error(err);}var result=void 0;try{result=JSON.stringify(ast,null,2);}catch(e){return error(e);}if(output){writeOut(result,output);}else{process.stdout.write(result);process.exit(0);}}function readCallback(err,data){if(err){return error(err);}(0,_index2.default)(data,parserCallback);}
+'use strict';var _index=require('../lib/index');var _index2=_interopRequireDefault(_index);var _fs=require('fs');var _path=require('path');function _interopRequireDefault(obj){return obj&&obj.__esModule?obj:{default:obj};}var aliases={o:'output',v:'version',h:'help',x:'stream'};var args=resolveArgs(process.argv.slice(2));var error=function error(err){console.error(err);process.exit(1);};var done=checkThen(function(){process.exit(0);});if(args['version']){console.log('sqlite-parser v1.0.0-beta2');process.exit(0);}if(args['help']||args._.length===0){console.log('Usage:\tsqlite-parser [infile]\n');console.log('Option\t\t\tAlias\tDescription');console.log('--output [outfile]\t-o\tWrite output to a file instead of stdout');console.log('--stream\t\t-x\tEnable streaming mode (default: infile >150kB)');console.log('--version\t\t-v\tGet current parser version');process.exit(0);}var streaming=args['stream'];var input=(0,_path.normalize)(args._[0]);var output=args['output'];if(output){output=(0,_path.normalize)(output);}(0,_fs.stat)(input,checkThen(function(_ref){var size=_ref.size;if(size/1000>=150){streaming=true;}var startStream=streaming?streamParser:standardParser;if(output){(function(){var outDir=(0,_path.dirname)(output);(0,_fs.stat)(outDir,checkThen(startStream,function(){(0,_fs.mkdir)(outDir,startStream);}));})();}else{startStream();}}));function resolveArgs(argv){var args={_:[]};var last=null;var isNewArg=function isNewArg(arg){return!arg||arg.indexOf('-')===0;};for(var i=0;i<argv.length;i+=1){var arg=argv[i];if(isNewArg(arg)){var cur=arg.indexOf('--')!==-1?arg.slice(2):aliases[arg.slice(1)];var peek=argv.length-1!==i?argv[i+1]:null;var peekNew=isNewArg(peek);args[cur]=peekNew?true:peek;if(!peekNew){i+=1;}}else{args._.push(arg);}}return args;}function checkThen(){var resCallback=arguments.length>0&&arguments[0]!==undefined?arguments[0]:done;var errCallback=arguments.length>1&&arguments[1]!==undefined?arguments[1]:error;return function(err,result){if(err){return errCallback(err);}resCallback(result);};}function streamParser(){var parserTransform=_index2.default.createParser();var singleNodeTransform=_index2.default.createStitcher();var readStream=(0,_fs.createReadStream)(input);var writeStream=output?(0,_fs.createWriteStream)(output):process.stdout;readStream.pipe(parserTransform);parserTransform.pipe(singleNodeTransform);singleNodeTransform.pipe(writeStream);parserTransform.on('error',error);singleNodeTransform.on('error',error);writeStream.on('finish',done);}function standardParser(){(0,_fs.readFile)(input,'utf8',checkThen(function(data){(0,_index2.default)(data,checkThen(function(ast){var result=void 0;try{result=JSON.stringify(ast,null,2);}catch(e){return error(e);}if(output){(0,_fs.writeFile)(output,result,checkThen(done));}else{process.stdout.write(result+'\n');done();}}));}));};

lib/streaming.js

@@ -3,15 +3,31 @@ import {
   stat,
   writeFile,
   mkdir,
-  readFile
+  readFile,
+  createReadStream,
+  createWriteStream
 } from 'fs';
 import {
   normalize,
   dirname
 } from 'path';
 
+const aliases = {
+  o: 'output',
+  v: 'version',
+  h: 'help',
+  x: 'stream'
+};
 const args = resolveArgs(process.argv.slice(2));
 
+const error = function (err) {
+  console.error(err);
+  process.exit(1);
+}
+const done = checkThen(function () {
+  process.exit(0);
+});
+
 if (args['version']) {
   console.log(`sqlite-parser v@@VERSION`);
   process.exit(0);
@@ -21,27 +37,38 @@ if (args['help'] || args._.length === 0) {
   console.log(`Usage:\tsqlite-parser [infile]\n`);
   console.log(`Option\t\t\tAlias\tDescription`);
   console.log(`--output [outfile]\t-o\tWrite output to a file instead of stdout`);
+  console.log(`--stream\t\t-x\tEnable streaming mode (default: infile >150kB)`);
   console.log(`--version\t\t-v\tGet current parser version`);
   process.exit(0);
 }
 
+let streaming = args['stream'];
 const input = normalize(args._[0]);
-let output = args['o'] || args['output'];
-
+let output = args['output'];
 if (output) {
   output = normalize(output);
 }
 
-stat(input, function startCallback(err) {
-  if (err) { return error(err); }
-  readFile(input, 'utf8', readCallback);
-});
+stat(input, checkThen(function ({ size }) {
+  // If the file size is above a 150kB limit, switch to streaming mode
+  if (size / 1000 >= 150) {
+    streaming = true;
+  }
+  const startStream = streaming ? streamParser : standardParser;
+  if (output) {
+    const outDir = dirname(output);
+    stat(outDir, checkThen(startStream, function () {
+      mkdir(outDir, startStream);
+    }));
+  } else {
+    startStream();
+  }
+}));
 
 function resolveArgs(argv) {
   const args = { _: [] };
   let last = null;
   const isNewArg = (arg) => !arg || arg.indexOf('-') === 0;
-  const aliases = { o: 'output', v: 'version', h: 'help' };
   for (let i = 0; i < argv.length; i += 1) {
     const arg = argv[i];
     if (isNewArg(arg)) {
@@ -57,53 +84,47 @@ function resolveArgs(argv) {
   return args;
 }
 
-function error(err) {
-  console.error(err.message);
-  process.exit(1);
+function checkThen(
+  resCallback = done,
+  errCallback = error
+) {
+  return function (err, result) {
+    if (err) { return errCallback(err); }
+    resCallback(result);
+  };
 }
 
-function writeOut(result, outPath) {
-  const outDir = dirname(outPath);
-
-  function writeCallback(err) {
-    if (err) { return error(err) }
-    process.exit(0);
-  }
+function streamParser() {
+  const parserTransform = parser.createParser();
+  const singleNodeTransform = parser.createStitcher();
+  const readStream = createReadStream(input);
+  const writeStream = output ? createWriteStream(output) : process.stdout;
 
-  function mkdirCallback(err) {
-    if (err) { return error(err) }
-    writeFile(outPath, result, writeCallback);
-  }
+  readStream.pipe(parserTransform);
+  parserTransform.pipe(singleNodeTransform);
+  singleNodeTransform.pipe(writeStream);
 
-  function statCallback(err) {
-    if (err) {
-      return mkdir(outDir, mkdirCallback);
-    }
-    mkdirCallback();
-  }
+  parserTransform.on('error', error);
+  singleNodeTransform.on('error', error);
 
-  stat(outDir, statCallback);
+  writeStream.on('finish', done);
 }
 
-function parserCallback(err, ast) {
-  if (err) { return error(err) }
-  let result;
-
-  try {
-    result = JSON.stringify(ast, null, 2);
-  } catch (e) {
-    return error(e);
-  }
-
-  if (output) {
-    writeOut(result, output);
-  } else {
-    process.stdout.write(result);
-    process.exit(0);
-  }
-}
-
-function readCallback(err, data) {
-  if (err) { return error(err) }
-  parser(data, parserCallback);
-}
+function standardParser() {
+  readFile(input, 'utf8', checkThen(function (data) {
+    parser(data, checkThen(function (ast) {
+      let result;
+      try {
+        result = JSON.stringify(ast, null, 2);
+      } catch (e) {
+        return error(e);
+      }
+      if (output) {
+        writeFile(output, result, checkThen(done));
+      } else {
+        process.stdout.write(`${result}\n`);
+        done();
+      }
+    }));
+  }));
+};

src/bin/sqlite-parser.js

@@ -75,7 +75,7 @@ export class SingleNodeTransform extends Transform {
   }
 
   _flush(callback) {
-    this.push(`\n  ]\n}`);
+    this.push(`\n  ]\n}\n`);
     callback();
   }
 }