Merge remote-tracking branch 'origin/large_files'

.istanbul.yml

@@ -0,0 +1,16 @@
+verbose: false
+instrumentation:
+  root: 'lib'
+  default-excludes: true
+  excludes: [ 'src', 'coverage' ]
+  embed-source: false
+  variable: __coverage__
+  compact: true
+  preserve-comments: false
+  complete-copy: false
+  save-baseline: false
+reporting:
+  print: summary
+  reports:
+    - lcovonly
+  dir: ./coverage

.travis.yml

@@ -9,6 +9,7 @@ node_js:
 services:
   - mysql
 before_script:
+  - mysql --version
   - mysql -e "create database IF NOT EXISTS knex;" -uroot
   - mysql -Dknex -uroot < test/fixtures/travis.sql
 script:

README.md

@@ -20,6 +20,7 @@ import seeder from 'knex-csv-seeder';
 exports.seed = seeder({
   table: 'users',
   file: '/path/to/users.csv',
+  // recordsPerQuery: 100,
   // encoding: 'utf8' default encoding
   // parser: {
   //   delimiter: ',',

package.json

@@ -8,7 +8,7 @@
     "README.md"
   ],
   "scripts": {
-    "test": "multi='dot=- mocha-lcov-reporter=lcov-report.lcov' ./node_modules/.bin/mocha test",
+    "test": "istanbul cover _mocha",
     "build": "./node_modules/.bin/babel -d lib -s inline src",
     "watch": "./node_modules/.bin/babel -d lib -s inline -w src"
   },
@@ -23,30 +23,22 @@
   },
   "homepage": "https://github.com/holyshared/knex-csv-seeder",
   "dependencies": {
-    "bluebird": "^3.0.5",
-    "csv-parse": "^1.0.0",
+    "bluebird": "^3.1.1",
+    "csv-parse": "^1.0.1",
     "iconv-lite": "^0.4.13",
     "knex": "^0.9.0",
     "lodash": "^3.10.1"
   },
   "devDependencies": {
-    "babel-cli": "^6.1.2",
-    "babel-preset-es2015": "^6.1.2",
-    "blanket": "^1.1.9",
+    "babel-cli": "^6.3.17",
+    "babel-preset-es2015": "^6.3.13",
     "codecov": "^1.0.1",
     "coffee-script": "^1.10.0",
-    "eslint": "^1.8.0",
-    "espower-coffee": "^1.0.0",
-    "mocha": "^2.3.3",
-    "mocha-lcov-reporter": "^1.0.0",
-    "mocha-multi": "^0.7.2",
-    "mysql": "^2.9.0",
-    "power-assert": "^1.1.0"
-  },
-  "config": {
-    "blanket": {
-      "pattern": "/lib/",
-      "data-cover-never": "/node_modules/"
-    }
+    "eslint": "^1.10.3",
+    "espower-coffee": "^1.0.1",
+    "istanbul": "^0.4.1",
+    "mocha": "^2.3.4",
+    "mysql": "^2.10.0",
+    "power-assert": "^1.2.0"
   }
 }

src/seeder.js

@@ -29,6 +29,12 @@ class KnexSeeder extends EventEmitter {
     this.headers = [];
     this.records = [];
     this.parser = null;
+    this.queue = null;
+    this.results = [];
+    this.onReadable = this.onReadable.bind(this);
+    this.onEnd = this.onEnd.bind(this);
+    this.onSucceeded = this.onSucceeded.bind(this);
+    this.onFailed = this.onFailed.bind(this);
   }
 
   static fromKnexClient(knex) {
@@ -41,6 +47,7 @@ class KnexSeeder extends EventEmitter {
       file: null,
       table: null,
       encoding: 'utf8',
+      recordsPerQuery: 100,
       parser: {
         delimiter: ',',
         quote: '"',
@@ -55,16 +62,19 @@ class KnexSeeder extends EventEmitter {
 
   generate(options) {
     this.opts = this.mergeOptions(options);
+
     this.parser = parse(this.opts.parser);
-    this.parser.on('readable', this.readable.bind(this) );
-    this.parser.on('end', this.end.bind(this) );
-    this.parser.on('error', this.error.bind(this) );
+    this.parser.on('readable', this.onReadable);
+    this.parser.on('end', this.onEnd);
+    this.parser.on('error', this.onFailed);
+
+    this.queue = Promise.bind(this).then( this.createCleanUpQueue() );
 
-    let csv = fs.createReadStream(this.opts.file);
-    csv.pipe( iconv.decodeStream(this.opts.encoding) ).pipe(this.parser);
+    this.csv = fs.createReadStream(this.opts.file);
+    this.csv.pipe( iconv.decodeStream(this.opts.encoding) ).pipe(this.parser);
   }
 
-  readable() {
+  onReadable() {
     let obj = {};
     let record = this.parser.read();
 
@@ -75,25 +85,58 @@ class KnexSeeder extends EventEmitter {
     if (this.parser.count <= 1) {
       this.headers = record;
     } else {
-      this.headers.forEach((column, i) => {
-        let val = record[i];
+      this.records.push( this.createObjectFrom(record) );
+    }
 
-        if (typeof val === 'string' && val.toLowerCase() === 'null') {
-          val = null;
-        }
-        obj[column] = val;
-      });
-      this.records.push(obj);
+    if (this.records.length < this.opts.recordsPerQuery) {
+      return;
     }
+
+    this.queue = this.queue.then( this.createBulkInsertQueue() );
+  }
+  onEnd() {
+    if (this.records.length > 0) {
+      this.queue = this.queue.then( this.createBulkInsertQueue() );
+    }
+    this.queue.then(() => {
+      return this.emit('end', this.results);
+    }).catch(this.onFailed);
+  }
+  createCleanUpQueue() {
+    return () => {
+      return this.knex(this.opts.table).del()
+        .then(this.onSucceeded)
+        .catch(this.onFailed);
+    };
+  }
+  createBulkInsertQueue() {
+    const records = this.records.splice(0, this.opts.recordsPerQuery);
+
+    return () => {
+      return this.knex(this.opts.table)
+        .insert(records)
+        .then(this.onSucceeded)
+        .catch(this.onFailed);
+    };
+  }
+  createObjectFrom(record) {
+    let obj = {};
+
+    this.headers.forEach((column, i) => {
+      let val = record[i];
+
+      if (typeof val === 'string' && val.toLowerCase() === 'null') {
+        val = null;
+      }
+      obj[column] = val;
+    });
+    return obj;
   }
-  end() {
-    const queues = [
-      this.knex(this.opts.table).del(),
-      this.knex(this.opts.table).insert(this.records)
-    ];
-    this.emit('end', Promise.join.apply(Promise, queues));
+  onSucceeded(res) {
+    this.results.push(res);
   }
-  error(err) {
+  onFailed(err) {
+    this.csv.unpipe();
     this.emit('error', err);
   }
 }