gasket.js

// run maps to lazy multistream

var execspawn = require('npm-execspawn')
var duplexify = require('duplexify')
var multistream = require('multistream')
var parallel = require('parallel-multistream')
var pumpify = require('pumpify')
var duplexify = require('duplexify')

var toStream = function(cmd) {
  if (cmd.pipe) return cmd // already a stream
  var proc = execspawn(cmd)
  return duplexify(proc.stdin, proc.stdout)
}

var runStream = function (runCommands) {
  var commands = runCommands.map(function (cmd) {
    return function () {
      // when multistream gets a function it creates the streams
      // lazily (like running commands in sequence)
      var s = toStream(cmd)
      if (s.end) s.end() // not writable
      return s
    }
  })
  return multistream(commands)
}

var forkStream = function (forkCommands) {
  var commands = forkCommands.map(function (cmd) {
      // no lazyness here since we are forking
      var s = toStream(cmd)
      if (s.end) s.end() // not writable
      return s
  })
  return parallel(commands)
}

var pipeStream = function(pipeCommands) {
  var commands = pipeCommands.map(function (cmd) {
    return toStream(cmd)
  })
  
  var pipe = pumpify(commands)
  pipe.end() // first not writable
  return pipe
}

// all of the above return streams (like a process)

// test.ds translated *roughly* into

// implicit runStream around the entire main pipeline
var pipeline = runStream([
  runStream([
    'echo hello',
    'echo world'
  ]),
  forkStream([
    'echo hello from fork',
    'echo world from fork'
  ]),
  pipeStream([
    'echo hello from pipe',
    'cat -'
  ])
])

pipeline.pipe(process.stdout)

test.ds

run
  echo hello
  echo world

fork
  echo hello from fork
  echo world from fork

pipe
  echo hello from pipe
  cat -