PrefixStream implemented with/out external dependencies

PrefixStreamFull.js

var util          = require('util');
var StringDecoder = require('string_decoder').StringDecoder;
var stream        = require('stream');

var Transform = stream.Transform ||
  require('readable-stream').Transform;

function PrefixStream(prefix, options) {
  if ( ! (this instanceof PrefixStream)) {
    return new PrefixStream(prefix, options);
  }
  Transform.call(this, options);

  this.prefix                    = prefix;
  this._writableState.objectMode = false;
  this._buffer                   = '';
  this._decoder                  = new StringDecoder('utf8');
}

util.inherits(PrefixStream, Transform);

PrefixStream.prototype._transform = function (chunk, encoding, cb) {
  this._buffer += this._decoder.write(chunk);
  // split on newlines
  var lines = this._buffer.split(/\r?\n/);
  // keep the last partial line buffered
  this._buffer = lines.pop();
  for (var l = 0; l < lines.length; l++) {
    // prefix and push out to consumer stream
    var line = util.format('%s%s\n', this.prefix, lines[l]);
    this.push(line);
  }
  cb();
};

PrefixStream.prototype._flush = function(cb) {
  // handle the leftovers
  var rem = this._buffer.trim();
  if (rem) {
    // prefix and push out to consumer stream
    var line = util.format('%s%s\n', this.prefix, rem);
    this.push(line);
  }
  cb();
};

module.exports = PrefixStream;

PrefixStreamSimple.js

var util    = require('util');
var stream  = require('stream');

var duplex  = require('duplexer');
var split   = require('split');
var through = require('through');

var PassThrough = stream.PassThrough ||
  require('readable-stream').PassThrough;

function PrefixStream (prefix) {
  if ( ! (this instanceof PrefixStream)) {
    return new PrefixStream(prefix);
  }

  prefix = prefix || '';

  this.inStream   = new PassThrough();
  this.outStream  = new PassThrough();

  var tr = through(function(line){
    line = util.format('%s%s\n', prefix, line);
    this.queue(line);
  });

  this.inStream
    .pipe(split())
    .pipe(tr)
    .pipe(this.outStream);

  return duplex(this.inStream, this.outStream);
};

module.exports = PrefixStream;