initial code

index.js

@@ -0,0 +1 @@
+module.exports = require('./lib');

lib/index.js

@@ -0,0 +1,26 @@
+var Async = require('async');
+var Fs = require('fs');
+var MongoClient = require('mongodb').MongoClient;
+
+var Loader = require('./loader');
+
+exports = module.exports = {
+  load: function (sourcePath, options, callback) {
+    Async.auto({
+      db: function (callback) {
+        console.log('opening db connection');
+        MongoClient.connect(options.mongoURI, callback);
+      },
+      subDirectory: function (callback) {
+        Fs.readdir(sourcePath, callback);
+      },
+      loadData: ['db', 'subDirectory', function (callback, results) {
+        Async.each(results.subDirectory, load(results.db), callback);
+      }],
+      cleanUp: ['db', 'loadData', function (callback, results) {
+        console.log('closing db connection');
+        results.db.close(callback);
+      }]
+    }, callback); 
+  }
+}

lib/loader.js

@@ -0,0 +1,31 @@
+var Csv = require('csv-streamify');
+var Es = require('event-stream');
+var Fs = require('fs');
+
+var Streams = require('./streams');
+
+var load = function (db, options) {
+  return function (directory, callback) {
+    var basePath = options.sourcePath + '/' + directory;
+    Async.waterfall([
+      function (callback) {
+        Fs.readdir(basePath, callback);
+      },
+      function (files, callback) {
+        console.log('loading ' + files.length + ' files from ' + directory);
+        Async.each(files, function (file, callback) {
+          Fs.createReadStream(basePath + '/' + file)
+            .pipe(Csv({objectMode: true, columns: true}))
+            .pipe(Streams.transform(directory))
+            .pipe(Streams.batch(200))
+            .pipe(Streams.insert(db).on('end', callback));
+        }, callback);
+      }
+    ], callback);
+  };
+};
+
+module.exports = {
+  load: load  
+};
+

lib/streams/batch.js

@@ -0,0 +1,25 @@
+var batch = function (batchSize) {
+  batchSize = batchSize || 1000;
+  var batch = [];
+
+  return Es.through(
+    function write (data) {
+      batch.push(data);
+      if (batch.length === batchSize) {
+        this.emit('data', batch);
+        batch = [];
+      }
+    },
+    function end () {
+      if (batch.length) {
+        this.emit('data', batch);
+        batch = [];
+      }
+      this.emit('end');
+    }
+  );
+};
+
+module.exports = {
+  batch: batch
+};

lib/streams/insert.js

@@ -0,0 +1,19 @@
+var insert = function (db) {
+  return Es.map(
+    function (data, callback) {
+      if (data.length) {
+        var bulk = db.collection('hnet').initializeUnorderedBulkOp();
+        data.forEach(function (doc) {
+          bulk.insert(doc);
+        });
+        bulk.execute(callback);
+      } else {
+        callback();
+      }
+    }
+  );
+};
+
+module.exports = {
+  insert: insert  
+};

lib/streams/transform.js

@@ -0,0 +1,11 @@
+var transform = function (directory) {
+  return Es.map(function (data, callback) {
+    data.siteRef = Mapping[directory];
+    data.epoch = parseInt((data.TheTime - 25569) * 86400) + 6 * 3600;
+    callback(null, data);
+  });
+};
+
+module.exports = {
+  transform: transform
+};