kd/lib/processes.coffee

# this is a work in progress.
# it's doing proper cleanup and management of child processes
# use it with care, not all edge cases are covered.
#
# author @devrim

{argv}              = require 'optimist'
traverse            = require 'traverse'
_                   = require 'underscore'
fs                  = require "fs"
os                  = require "os"
hat                 = require 'hat'
{exec,spawn,fork}   = require "child_process"
{EventEmitter}      = require "events"


################################
### HANDLE SIGNALS and KILLS ###
################################

# guaranteed death.
process.on 'SIGTERM', =>
  console.log "going down"
  process.exit(0)

class Processes extends EventEmitter
  
  constructor:(options={})->
    {main, processContainerDir} = options
    @list          = {}
    @child         = null
    @workerProcess = {}
    @childProcessesContainerDir = processContainerDir or "/tmp/kd/"

    try
      fs.mkdirSync @childProcessesContainerDir
    catch e
      # do nothing, probably dir is already there.

    if main
      previouschildProcessesContainerDir = @childProcessesContainerDir+"."+Date.now()
      fs.renameSync @childProcessesContainerDir, previouschildProcessesContainerDir
      fs.mkdirSync @childProcessesContainerDir
      exec "rm -rf #{previouschildProcessesContainerDir}",(err,stdout,stderr)->
        if err
          console.log "couldn't delete previous process tree at #{previouschildProcessesContainerDir}"
        
  fork: (options,callback) ->
    options.fork = yes
    @run options,callback, (cb) =>
      callback cb

  spawn: (options,callback)->
    options.spawn = yes
    @run options,callback

  exec: (cmd,callback)->
    @spawn
      name  : Date.now()
      cmd   : cmd
      restart : no
      stdout : process.stdout
      stderr : process.stderr
      onExit  : callback

  run: (options,callback) ->
    {name, restartTimeout, verbose} = options

    options.name            ?= Date.now()
    options.restartTimeout  ?= 1000
    options.verbose         ?= no
    @child = @startProcess options, callback
    child = @populateChildObject options, @child
    callback false, child 

  getDate: ->
    d = new Date()
    pad = (n) ->
      (if n < 10 then "0" + n else n)
    d.getUTCFullYear() + "-" + pad(d.getUTCMonth() + 1) + "-" + pad(d.getUTCDate()) + "T" + pad(d.getUTCHours()) + ":" + pad(d.getUTCMinutes()) + ":" + pad(d.getUTCSeconds()) + "Z"

  startProcess: (options, callback) ->
    {name, args, opts, cmd, stdout, stderr, verbose} = options

    if "string" is typeof options.cmd
      cmdA = cmd.split(" ")
    else if Array.isArray cmd
      cmdA = cmd
    else
      console.log e = "[PROCESSES][ERROR][#{name}] Can't start without a valid cmd param." if verbose
      return callback e

    if options.fork
      @child = fork cmdA[0],cmdA[1...],opts
    else
      # @child = spawn cmdA[0],cmdA[1...]
      console.log ">>>", cmd, args
      @child = spawn cmd, args # cmdA is split wrongly when quotation marks are involved
      @child.stdout.pipe stdout if stdout
      @child.stderr.pipe stderr if stderr

    return @child


  populateChildObject: (options, @child) ->
    {name,onMessage,modulePath,args,restart,restartTimeout,cmd,log,onExit,stdout,stderr,verbose,die} = options
    onExit ?= ->
    console.log "[PROCESSES][#{name}] started the process: #{name} with pid: #{@child.pid}" if verbose
    @createPidFile @child.pid,name

    childObj =
      name    : name
      process : @child
      pid     : @child.pid
      verbose : verbose
      startedAt: Date.now()

    if cmd
      childObj.cmd = cmd
    else if modulePath
      childObj.fork = {modulePath,args,options}

    if @list[name]
      d = Date.now()
      console.log "[WARN][Processes] process name '#{name}' not unique. to be able to kill it properly it is now: #{name}-#{d}" if verbose
      name += "-"+d

    @list[name] = childObj

    if die?.after
      console.log "[PROCESSES][#{name}]i will kill this process:#{name} every #{die.after}msecs" if verbose
      setTimeout =>
        console.log "[PROCESSES][#{name}]it's time you die process:#{name}" if verbose
        @kill name
      ,die.after

    @child.stdout.on 'data', (data)->
      process.stdout.write("[PROCESSES][#{name}]" + data)
    
    @child.stderr.on 'data', (data)->
      console.log("[PROCESSES][ERROR][#{name}]" + data)

    @child.on 'exit', ()=>
      console.log "[PROCESSES][#{name}] did exit." if verbose
      @emit "exit",@child.pid,name
      delete @list[name]
      if options.kontrol?.enabled
        @workerProcess[name].stopReporter()
        @workerProcess[name].connection.end()
        delete @workerProcess[name]

    @child.on 'close', () =>
      console.log "[PROCESSES][#{name}] did close." if verbose
      @emit "close",@child.pid,name
      onExit @child.pid,name
      if restart is yes
        restartTimeout = 1 unless restartTimeout
        setTimeout =>
          @run options
        ,restartTimeout
        console.log "[PROCESSES][#{name}] restarting the process in #{restartTimeout} msecs." if verbose

    @child.on 'message',(msg) =>
      onMessage msg if "function" is typeof onMessage
      if msg['process-monitor']
        delete msg['process-monitor']
        if options.kontrol?.enabled
          msg['uuid'] = @workerProcess[name].worker.uuid
          @workerProcess[name].fetchWorkerExchange (workerExchange) =>
            @workerProcess[name].worker.message.command = "ack"
            workerExchange.publish "input.worker", {monitor: msg},
              appId: "#{@workerProcess[name].worker.uuid}"
    return childObj

  createPidFile : (pid,name)->
    console.log "creating pid file"
    fs.writeFileSync "#{@childProcessesContainerDir}/#{process.pid}.#{pid}",""
    console.log "pid file created"

  get : (name)->
    if @list[name]
      return @list[name].process
    else
      console.log "[ERROR][Processes] #{name} Can't be found."

  fetchChildren: (parentPid,callback)->
    findChildren = (arr, pid) ->
      findChildrenRec = (pid,root=yes)->
        c={}
        pid = ""+pid
        i = -1
        arr.forEach (item)->
          [parent,child] = item.split(".")
          if parent is pid
            c[child] = findChildrenRec child,no
        return c

      arr = traverse(findChildrenRec pid).paths()
      arr.forEach (group) -> group.reverse()
      arr = _.sortBy arr,(a)-> a.length*-1
      return _.uniq _.flatten arr,yes

    fs.readdir @childProcessesContainerDir,(err,list)->
      if err
        callback err
      else
        callback null,findChildren list,parentPid

  killAllChildren: (parentPid, callback) ->
    @fetchChildren parentPid,(err,orderedKillList)->
      console.log "ordered kill list:"+orderedKillList
      unless err
        pidPair = []
        orderedKillList.forEach (pid)->
          console.log "killing children: #{pid}"
          try
            pidPair.push pid
            process.kill(pid*1)
            if pidPair.length is 2
              pidPair.reverse()
              deleteFile = pidPair.join(".")
              console.log "deleting"+deleteFile
              fs.unlinkSync deleteFile
              pidPair.shift()
          catch e
            console.log "child process #{pid} did not exist. not killed by me."
        callback null

  sendMessage: (name,msg)->
    if @list[name]?.process?.send
      @list[name].process.send msg

  kill: (name,callback=->)->
    unless @list[name]
      console.log "[PROCESSES][#{name}] cant kill #{name}, couldn't find it."
    else
      pid = @list[name].pid
      console.log "[PROCESSES][#{name}] killing parent: #{pid}"
      @killAllChildren pid,(err,res)->
        console.log "[PROCESSES][#{name}] killing #{name} #{pid}"
        process.kill pid

module.exports = Processes