Merge pull request #34 from CfABrigadePhiladelphia/import_pipeline

import files with importPipeline

Author: CameronSima

index.js

@@ -1,6 +1,6 @@
 var level = require('level')
 var hypercore = require('hypercore')
-var createImportStream = require('./lib/import.js')
+var createImportPipeline = require('./lib/import.js')
 
 module.exports = Jawn
 
@@ -11,6 +11,6 @@ function Jawn (opts) {
   this.db = this.core.db
 }
 
-Jawn.prototype.createImportStream = function (opts) {
-  return createImportStream(this, opts)
+Jawn.prototype.createImportPipeline = function (opts, callback) {
+  return createImportPipeline(this, opts, callback)
 }

lib/import.js

@@ -1,32 +1,30 @@
-var pump = require('pump')
+var miss = require('mississippi')
 var through = require('through2')
+var parseInputStream = require('parse-input-stream')
 
-module.exports = importStream
+module.exports = importPipeline
 
-function importStream (jawn, opts) {
+// Returns the parser at the beginning of the pipeline
+// When the pipeline finishes, it calls `callback(feedId, err)`.
+//   If there were no errors, err will be undefined.
+function importPipeline (jawn, opts, callback) {
   if (!opts) opts = {}
   var writeStream = jawn.core.createWriteStream(opts)
-  var stream = pump(parseInputStream(opts), writeStream, function done (err) {
-    stream.id = writeStream.id
-    console.log(err)
-  })
-  return stream
-}
+  var parser = parseInputStream(opts)
+  var transform = through.obj(stringifyData, end)
 
-// Transform input CSV, JSON, etc to json objects that will be written as blocks in our hypercore feed
-// TODO: Implement this! See https://github.com/CfABrigadePhiladelphia/jawn/issues/32
-// This is the functionality that parse-input-stream aims to support.
-// If you can get that module to work, feel free to use it!
-function parseInputStream (opts) {
-  var transformStream = through(write, end)
-  return transformStream
+  miss.pipe(parser, transform, writeStream, function done (err) {
+    callback(err, writeStream.id)
+  })
 
-  function write (buffer, encoding, next) {
-    // transform input CSV, JSON, etc. here ...
+  function stringifyData (data, enc, next) {
+    this.push(JSON.stringify(data))
     next()
   }
 
   function end (done) {
     done()
   }
+
+  return parser
 }

package.json

@@ -24,9 +24,9 @@
     "fs": "0.0.2",
     "hypercore": "~1.8.0",
     "level": "~1.4.0",
+    "mississippi": "^1.2.0",
     "parse-input-stream": "^1.0.1",
     "path": "~0.12.7",
-    "pump": "^1.0.1",
     "tape": "~4.5.1",
     "through2": "^2.0.1"
   },

test/data/dummy.json

@@ -0,0 +1,3 @@
+{"foo":"bar","name":"josie","age":"35"}
+{"foo":"baz","name":"eloise","age":"71"}
+{"foo":"baz","name":"francoise","age":"5"}

test/import.js

@@ -1,39 +1,61 @@
 var test = require('tape')
+var fs = require('fs')
+var path = require('path')
 var Jawn = require('../')
 var memdb = require('memdb')
 
 test('import json to jawn', function (t) {
   var jawn = freshJawn()
-  var importStream = jawn.createImportStream({'format': 'json'})
-  // Imitate the stream that would come from reading sample.csv
-  // importStream should parse the CSV correctly, identifying the first row as headers
-  // This is the same as doing
-  //    var data = fs.createReadStream('./test/data/sample.csv')
-  //    data.pipe(importStream)
-  // except the writes are being performed synchronously/inline so we can call importStream.end() after writing the contents into it.
-  importStream.write('{foo: "bar", name: "josie"}')
-  importStream.write('{foo: "baz", name: "eloise"}')
-  importStream.write('{foo: "baz", name: "francoise"}')
+  importFromFile(jawn, 'dummy.json', {'format': 'json'}, verify)
+  var expected = [
+    '{"foo":"bar","name":"josie","age":"35"}',
+    '{"foo":"baz","name":"eloise","age":"71"}',
+    '{"foo":"baz","name":"francoise","age":"5"}'
+  ]
+  function verify (err, feedId) {
+    if (err) { console.log(err) }
+    var rs = jawn.core.createReadStream(feedId)
+    rs.on('data', function (block) {
+      t.same(block.toString(), expected.shift(), 'block matches imported line')
+    })
+    t.same(jawn.core.get(feedId).blocks, 3, 'correct number of blocks returned')
+    t.end()
+  }
+})
 
-  // This should be expecting JSON objects, not strings.
-  // temporarily expecting strings in order to hand off the code as-is
+test('import csv to jawn', function (t) {
+  var jawn = freshJawn()
+  importFromFile(jawn, 'sample.csv', {'format': 'csv'}, verify)
   var expected = [
-    '{foo: "bar", name: "josie"}',
-    '{foo: "baz", name: "eloise"}',
-    '{foo: "baz", name: "francoise"}'
+    '{"Type of Experience":"Writing software in any programming language","Little/No Experience":"1","Some Experience":"5","Very Familiar":"4"}',
+    '{"Type of Experience":"Frontend Web Development","Little/No Experience":"4","Some Experience":"3","Very Familiar":"3"}',
+    '{"Type of Experience":"Server-side (“backend”) Web Development","Little/No Experience":"4","Some Experience":"4","Very Familiar":"2"}',
+    '{"Type of Experience":"Using Git to track changes and share code (add, commit, push, pull)","Little/No Experience":"2","Some Experience":"5","Very Familiar":"3"}'
   ]
 
-  importStream.end(function () {
-    var feedId = importStream.id.toString('hex')
+  function verify (err, feedId) {
+    if (err) { console.log(err) }
     var rs = jawn.core.createReadStream(feedId)
     rs.on('data', function (block) {
       t.same(block.toString(), expected.shift(), 'block matches imported line')
     })
-    t.same(jawn.core.get(feedId).blocks, 3, 'correct number of blocks returned')
+    t.same(jawn.core.get(feedId).blocks, 4, 'correct number of blocks returned')
     t.end()
-  })
+  }
 })
 
+// helpers
+
+function fixture (name) {
+  return path.join(__dirname, 'data', name)
+}
+
 function freshJawn () {
   return new Jawn({db: memdb()})
 }
+
+function importFromFile (jawn, file, opts, callback) {
+  var importPipeline = jawn.createImportPipeline(opts, callback)
+  var data = fs.createReadStream(fixture(file))
+  data.pipe(importPipeline)
+}