app.coffee

###
Created for http://stackoverflow.com/questions/34982953/child-process-stream-backpressure
Please pardon the CoffeeScript, but I couldn't stand to extend stream.Transform in native JavaScript.
###
fs = require("fs")
execStream = require("exec-stream")
brake = require("brake")

file = fs.createWriteStream("tmp.txt")

class Double extends require("stream").Transform
  _transform: (chunk, enc, cb) ->
    @_last ?= Date.now()
    @_called ?= []
    @_called.push Date.now() - @_last
    @_last = Date.now()
    @push chunk.toString() + chunk.toString()
    cb()

class UpperCase extends require("stream").Transform
  _transform: (chunk, enc, cb) ->
    @push chunk.toString().toUpperCase()
    cb()

sum = (nums) ->
  o = 0
  o += i for i in nums
  o

doTest = (size) ->

  transform1 = new Double()
  transform2 = new UpperCase()
  transform3 = new Double()

  execStream("dd", ["if=/dev/urandom", "bs=1024", "count=1"])
    .pipe(transform1)
    .pipe(transform2)
    .pipe(brake(size))
    .pipe(transform3)
    .pipe(file)

  file.on "finish", ->
    fs.stat "tmp.txt", (err, stats) ->
      throw err if err
      called1 = transform1._called
      averagePreBrake = sum(called1) / called1.length
      called2 = transform3._called
      averagePostBrake = sum(called2) / called2.length
      console.log """
        Generated with brake(#{size}): #{stats.size}
        Average time between transformations pre-brake:  #{averagePreBrake}ms
        Average time between transformations post-brake: #{averagePostBrake}ms
      """

doTest 1024
doTest 256

output.png