Designing and Using Pipelines with Node.js Streams, Browserify, and Gulp

Background: The article follows two previous posts about stream basics and internals, and focuses on applying pipelines in program design.

Pipeline concept: a pipeline is a composition of Stream objects connected via source.pipe(bold).pipe(red).pipe(dest). It can be encapsulated as a single Stream, allowing reuse and modification.

var pipeline = new Duplex()
var streams = pipeline._streams = [bold, red]
// _write: write to the first stream
pipeline._write = function (buf, enc, next) {
  streams[0].write(buf, enc, next)
}
// _read: read from the last stream
pipeline._read = function () {
  var buf, reads = 0
  var r = streams[streams.length - 1]
  while ((buf = r.read()) !== null) {
    pipeline.push(buf)
    reads++
  }
  if (reads === 0) {
    r.once('readable', function () { pipeline._read() })
  }
}
// combine streams (equivalent to bold.pipe(red))
streams.reduce(function (r, next) { r.pipe(next); return next })

Modifications such as inserting a new underline stream or replacing red with green are shown using pipeline._streams.splice, unpipe, and push/pop operations.

Utility libraries stream-splicer and labeled-stream-splicer provide array‑like methods (splice, push, pop) and named access for pipelines, enabling flexible reconfiguration.

Browserify uses a pipeline of Transform streams to process module objects (rows). The stages include record, deps, json, unbom, syntax, sort, dedupe, label, emit-deps, debug, pack, and wrap. Plugins can modify any stage via b.pipeline.get('stage').push(...).

Gulp’s core logic separates task scheduling (orchestrator) and file processing (vinyl‑fs). A typical Gulp task creates a pipeline from gulp.src to gulp.dest with intermediate Transform plugins, e.g.:

gulp.task('scripts', ['clean'], function () {
  return gulp.src(paths.scripts)
    .pipe(sourcemaps.init())
    .pipe(coffee())
    .pipe(uglify())
    .pipe(concat('all.min.js'))
    .pipe(sourcemaps.write())
    .pipe(gulp.dest('build/js'));
});

Comparison: both Browserify and Gulp rely on pipelines for their plugin mechanisms, but Browserify provides a predefined pipeline for module rows, while Gulp offers a flexible pipeline for vinyl file streams, leaving the composition entirely to the user.

Example: the ezchangelog tool builds a pipeline to parse a git‑log stream into markdown. The pipeline consists of a “parse” stage (split, commit, tag, url) and a “format” stage (markdownify). The implementation exposes a Changelog class, allows plugins, and is used as source.pipe(changelog()).pipe(dest).

function Changelog(opts) {
  this._options = opts || {}
  this.pipeline = splicer.obj([
    'parse', [
      'split', split(),
      'commit', commit(),
      'tag', tag(),
      'url', url({ baseUrl: opts.baseUrl })
    ],
    'format', [
      'markdownify', markdownify()
    ]
  ])
  ;[].concat(opts.plugin).filter(Boolean).forEach(function (p) { this.plugin(p) }, this)
}
Changelog.prototype.plugin = function (p, opts) {
  if (Array.isArray(p)) { opts = p[1]; p = p[0] }
  p(this, opts); return this
}

A custom plugin can add an author parser and replace the markdown formatter with a custom one that outputs SHA‑1, date, and author:

function customFormatter(c) {
  c.pipeline.get('parse').push(through.obj(function (ci, enc, next) {
    ci.committer.author = ci.headers[0][1].split(/\s+/)[0]
    next(null, ci)
  }))
  c.pipeline.get('format').splice('markdownify', 1, through.obj(function (ci, enc, next) {
    var sha1 = '`' + ci.commit.short + '`'
    var date = ci.committer.date.toISOString().slice(0,10)
    next(null, '* ' + sha1 + ' ' + date + ' @' + ci.committer.author + '
')
  }))
}

References to related GitHub repositories are listed at the end of the article.