node package manager
We need your input. Help make JavaScript better: Take the 2017 JavaScript Ecosystem survey »

pipe-segment

node-pipe-segment

Think of a pipe-segment as a bundle of streams, that do one thing together. See the discussion here: https://github.com/jbenet/random-ideas/issues/24 . Pipe segments are designed to be joined together into more complicated pipelines. It's similar in spirit to substack/labeled-stream-splicer, but rather than making one big pipeline, it uses segments that expose different streams for different things.

Inherit from this module and follow the pattern to make reusable components that can be joined together.

Install

npm install pipe-segment

Examples

Integrity Checksum pipe segment

Let's make a stream that wraps raw data with checksums (into a checksum-buffer). This is just a simple transform stream.

var multihash = require('multihashes')
var CkBuffer = require('checksum-buffer')
var through2 = require('through2')
 
function cksumWrap(func) {
  func = multihash.coerceCode(func) // ensure valid func code 
  return through2.obj(cksumAdd)
 
  function cksumAdd(buf, enc, next) {
    return CkBuffer(buf, func).buffer
  }
}

And one that unwraps a checksum-buffer. This is just a simple transform stream again.

var CkBuffer = require('checksum-buffer')
var through2 = require('through2')
 
function cksumUnwrap() { // dont need func 
  return through2.obj(cksumRemove)
 
  function cksumRemove(ckbuf, enc, next) {
    return CkBuffer(ckbuf).data()
  }
}

Now, let's make a segment that checks the checksum on a checksum-buffer. A segment is like a bundle of streams that are all related. This one is basically a pipe-segment-filter, so its stream interfaces will be input, output, and failed.

var CkBuffer = require('checksum-buffer')
var multihash = require('multihashes')
var filterSegment = require('pipe-segment-filter')
// filterSegment is like a T pipe segment. 
// it returns a segment object, with three streams: 
// - segment.input 
// - segment.output 
// - segment.filtered 
 
function cksumCheckSegment(func) {
  // func = optional check for a specific multihash function 
  if (func) func = multihash.coerceCode(func)
  return filterSegment(cksumCheck)
 
  function cksumCheck(buf) {
    var ckbuf = CkBuffer(buf)
    if (func && ckbuf.mhparams.code !== func)
      return false
    return ckbuf.check()
  }
}

Now, let's make a checksum integrity checked pipe-segment.

// we'll reuse the two functions above 
var segment = require('pipe-segment') // makes pipe segments 
var duplexer2 = require('duplexer2')
 
function integrityTransportSegment(cksumFunc) {
  //  raw   ----------- wrap ------->  cksum 
  //  side  <--- unwrap --- check ---  side 
 
  // make the internal streams 
  var check = cksumCheckSegment(cksumFunc)
  var wrap = cksumWrap(cksumFunc)
  var unwrap = cksumUnwrap()
 
  // wire up the internal streams 
  check.output.pipe(unwrap)
 
  // make duplex interfaces for each "side" of the streams 
  var side1 = duplexer2(wrap, unwrap) // read from unwrap, write to wrap 
  var side2 = duplexer2(check.input, wrap)
  // read from wrap, write to check.input 
 
  return segment({
    'raw': side1,
    'checksum': side2,
    'filtered': check.filtered,
  })
}

Boom, now we have a T pipe that wraps/unwraps/checks multihash checksums.