align takes two objectMode streams and a sequence key and will create a stream
aligning records from each stream resulting in a stream that emits record doublets
when the sequence key exactly matches. IMPORTANT: This only works on ordered streams!
@param  {String} key The sequence key the two streams are storted by and will be joined by.
@param  {Stream} leftStream  An objectMode stream ordered by `key`
@param  {Stream} rightStream An objectMode stream ordered by `key`
@return {Stream}             A doublet stream containing [leftRecord, rightRecord], [leftRecord, null], or [null, rightRecord]
function align(key, leftStream, rightStream, epsilon) {
  // TODO probably need to add stream options support for thigns like high water mark
  var aligner = new Aligner(key, {}, epsilon)

  var l = left()
  var r = right()

  var ended = 0
  l.on("end", function () {
    l.unpipe(aligner)
    if (++ended == 2) aligner.end()
  })
  r.on("end", function () {
    r.unpipe(aligner)
    if (++ended == 2) aligner.end()
  })

  leftStream.pipe(l).pipe(aligner, {end: false})
  rightStream.pipe(r).pipe(aligner, {end: false})

  return aligner
}