Working on streams

package.json

@@ -28,6 +28,7 @@
     "front-matter": "^2.0.6",
     "fs-date-parser": "^1.0.0",
     "good-thumbs": "^1.6.0",
-    "slug": "^0.9.1"
+    "slug": "^0.9.1",
+    "through2": "^2.0.1"
   }
 }

processData.js

@@ -5,6 +5,7 @@ var path = require('path')
 var slug = require('slug')
 var frontMatter = require('front-matter')
 var dateParser = require('fs-date-parser')
+var Readable = require('stream').Readable
 
 function preparePathForSlug (pth) {
   var pos = pth.search(/\.[^.]+$/)
@@ -82,12 +83,8 @@ function postCompiler (callback, err, compilerContext) {
   callback(null, data)
 }
 
-module.exports = function processData (raw, options, callback) {
-  if (typeof options === 'function') {
-    callback = options
-    options = {}
-  }
-  var fm = frontMatter(raw)
+function processString (rawString, options, callback) {
+  var fm = frontMatter(rawString)
   gatherDefaults(fm.attributes, options, function (ignoreError, data) {
     data.body = fm.body
     var compilerContext = {
@@ -107,3 +104,34 @@ module.exports = function processData (raw, options, callback) {
     postCompiler(callback, null, compilerContext)
   })
 }
+
+module.exports = function processData (raw, options, callback) {
+  if (typeof options === 'function') {
+    callback = options
+    options = {}
+  }
+  if (typeof raw === 'function') {
+    raw = raw(options)
+  }
+  if (raw instanceof Readable) {
+    var _onceDone = false
+    var once = function (err, data) {
+      if (_onceDone) return
+
+      _onceDone = true
+      callback(err, data)
+    }
+    raw.on('error', once)
+    var stream = raw.pipe(require('./transform')(options))
+    stream.on('data', once.bind(null, null))
+    stream.on('error', once)
+    return
+  }
+  if (raw === null || raw === undefined) {
+    return setImmediate(callback.bind(null, new Error('No data given to process.')))
+  }
+  if (typeof raw !== 'string') {
+    raw = raw.toString()
+  }
+  return processString(raw, options, callback)
+}

processFile.js

@@ -9,10 +9,5 @@ module.exports = function (filepath, options, callback) {
     options = {}
   }
   options.filepath = filepath
-  fs.readFile(filepath, 'utf-8', function (err, body) {
-    if (err) {
-      return callback(err)
-    }
-    processData(body, options, callback)
-  })
+  processData(fs.createReadStream(filepath), options, callback)
 }

test/processData.js

@@ -71,6 +71,30 @@ describe('Using default options', function (it) {
     })
   })
 })
+describe('Should accept uncommon input', function (it) {
+  it('should resolve the function', function (t) {
+    processData(function () { return 'hello' }, {}, function (ignore, data) {
+      t.equal(data.body, 'hello')
+      t.end()
+    })
+  })
+  it('should not allow null or undefined', function (t) {
+    processData(null, {}, function (error, data) {
+      t.equal(error.message, 'No data given to process.')
+      t.end()
+    })
+  })
+  it('should stringify an object', function (t) {
+    processData({
+      toString: function () {
+        return 'hello'
+      }
+    }, {}, function (ignore, data) {
+      t.equal(data.body, 'hello')
+      t.end()
+    })
+  })
+})
 describe('Capabilities to do custom links', function (it) {
   it('should be a simple function callback', function (t) {
     processData('', {data: {slug: 'fancy'}, linkIt: function (link) { return '/' + link }}, function (ignore, data) {

transform.js

@@ -0,0 +1,24 @@
+'use strict'
+
+var processData = require('./processData')
+var through2 = require('through2')
+
+module.exports = function (options) {
+  var buffers = []
+  return through2.obj(
+    function (chunk, enc, cb) {
+      buffers.push(chunk)
+      cb(null, null)
+    },
+    function (cb) {
+      var buffer = Buffer.concat(buffers)
+      var next = function (err, argument) {
+        if (argument) {
+          this.push(argument)
+        }
+        cb(err)
+      }
+      processData(buffer, options, next.bind(this))
+    }
+  )
+}