a transform stream

index.js

var stream = require('stream')

module.exports = function(opts) {

  opts = opts || { objectMode: true }

  return function(fn) {
    var s = new stream.Transform(opts)

    s._transform = function (chunk, encoding, done) {
      var that = this;
      var fin = function(data) {
        if (typeof data !== 'undefined') {
          that.push(data)
        }
        done()
      }
      var ret = fn.call(this, chunk, fin)
      if(ret) {
        this.push(ret)
        done()
      }
    }

    s._flush = function (done) {
      var ret = fn.call(this, null, done)
      if(ret || ret === null) {
        done()
      }
    }
    return s
  }
}

package.json

{
  "name": "txs",
  "version": "1.0.0",
  "description": "transform stream",
  "main": "index.js",
  "scripts": {
    "test": "node ./test.js"
  },
  "keywords": [
    "transform",
    "through",
    "throughstream",
    "throughstreams",
    "streams",
    "streaming"
  ],
  "author": "hij1nx",
  "license": "MIT",
  "dependencies": {},
  "repository": {
    "type": "git",
    "url": "https://gist.github.com/hij1nx/9022855"
  },
  "devDependencies": {
    "better-assert": "~1.0.0"
  }
}

test.js

var ASSERT = require('assert').ok
var txs = require('./index')()
var fs = require('fs')
var Stream = require('stream').Stream

var s1 = txs(function(obj, done) {
	return obj
})

var s2 = txs(function(obj, done) {
	done(obj)
})