Jump To …


client              = require './client'
fs                  = require 'fs'
{exists}            = require 'path'
{pause, isFunction} = require './util'
{debug}             = require './util'
{LineBuffer}        = require './util'
{spawn, exec}       = require 'child_process'
{EventEmitter}      = require 'events'
{Stream}            = require 'net'

packageBin = fs.realpathSync "#{__dirname}/../../bin"
packageLib = fs.realpathSync "#{__dirname}/.."

Process manages a single Ruby worker process.

A Process requires a path to a rackup config file (config.ru).


You can set a idle time so the process dies after a specified amount of milliseconds.

var ms = 15 * 60 * 1000;
process.createProcess("/path/to/app/config.ru", { idle: ms });

A Process has 5 states:

null: Not started or dead

spawning: Is booting and can't accept a connection yet

ready: Is ready to accept a connection and handle a request

busy: Is currently handling a request, not ready

quitting: Was set a kill signal and is shutting down

Anytime a process changes states, an event is fired with the new state name. When the process becomes ready, a ready is fired.

Other events:

Event: 'error'

function (exception) { }

Emitted when an error occurs.

Event: 'spawn'

function (exception) { }

Emitted when the process moves from spawning to ready state.

Event: 'exit'

function () { }

Emitted when the Process terminates.

Event: 'idle'

function () { }

Emitted when the Process times out because of inactivity.

exports.Process = class Process extends EventEmitter
  constructor: (@config, options) ->
    self = @
    @id = Math.floor Math.random() * 1000

    options ?= {}
    @idle  = options.idle
    @cwd   = options.cwd
    @env   = options.env ? {}

Set initial state to null

    @state = null

    @_connectionQueue = []
    @_activeConnection = null

    raiseConfigError = ->
      self.emit 'error', new Error "configuration \"#{@config}\" doesn't exist"

Raise an exception unless config exists

    if @config?
      exists @config, (ok) ->
        raiseConfigError() if !ok

    @on 'ready', ->

    @on 'error', (error) ->
      callback = self._activeConnection
      self._activeConnection = null

      if callback
        callback error
      else if self.listeners('error').length <= 1
        throw error

Push back the idle time everytime a request is handled

    @on 'busy', ->

  spawn: ->

Do nothing if the process is already started

    return if @state

    debug "spawning process"

Change start to spawning and fire an event

    @changeState 'spawning'

Generate a random sock path

    @sockPath = "#{tmpFile()}.sock"

Copy process environment

    env = {}
    for key, value of process.env
      env[key] = value

    for key, value of @env
      env[key] = value

    env['PATH']    = "#{packageBin}:#{env['PATH']}"
    env['RUBYLIB'] = "#{packageLib}:#{env['RUBYLIB']}"

    @heartbeat = new Stream

    @heartbeat.on 'connect', =>
      debug "process spawned"
      @emit 'spawn'

    @heartbeat.on 'data', (data) =>
      if "#{@child.pid}\n" is data.toString()
        @changeState 'ready'
          exception   = JSON.parse data
          error       = new Error exception.message
          error.name  = exception.name
          error.stack = exception.stack

          debug "heartbeat error", error
          @emit 'error', error
        catch e
          debug "heartbeat error", e
          @emit 'error', new Error "unknown process error"

    tryConnect @heartbeat, @sockPath, (err) =>
      if err and out
        @emit 'error', new Error out
      else if err
        @emit 'error', err

Spawn a Ruby server connecting to our @sockPath

    @child = spawn "nack_worker", [@config, @sockPath],
      cwd: @cwd
      env: env

Expose stdout and stderr on Process

    @stdout = @child.stdout
    @stderr = @child.stderr

    out = null
    logData = (data) ->
      out ?= ""
      out += data.toString()

    @stdout.on 'data', logData
    @stderr.on 'data', logData

    @on 'spawn', =>
      out = null
      @stdout.removeListener 'data', logData
      @stderr.removeListener 'data', logData

When the child process exists, clear out state and emit exit

    @child.on 'exit', (code, signal) =>
      debug "process exited"

      @heartbeat.destroy() if @heartbeat

      @state = @sockPath = null
      @child = @heartbeat = null
      @stdout = @stderr = null

      @_connectionQueue = []
      @_activeConnection = null

      @emit 'exit'


Change the current state and fire a corresponding event

  changeState: (state) ->
    self = this
    if @state != state
      @state = state

State change events are always asynchronous

      process.nextTick -> self.emit state

Clear current timeout handler.

  clearTimeout: ->
    if @_timeoutId
      clearTimeout @_timeoutId

Defer the current idle timer.

  deferTimeout: ->
    self = this
    if @idle

Clear the current timer


      callback = ->
        self.emit 'idle'

Register a new timer

      @_timeoutId = setTimeout callback, @idle

  _processConnections: ->
    self = @

    unless @_activeConnection
      @_activeConnection = @_connectionQueue.shift()

    if @_activeConnection and @state is 'ready'

Immediately flag process as busy

      @changeState 'busy'

Create a connection to our sock path

      connection = client.createConnection @sockPath

When the connection closes, change the state back to ready.

      connection.on 'close', ->
        self._activeConnection = null
        self.changeState 'ready'

      @_activeConnection null, connection

Create a new Client connection

  createConnection: (callback) ->
    @_connectionQueue.push callback

Proxies a http.ServerRequest and http.ServerResponse to the process

  proxyRequest: (req, res, args...) ->
    debug "proxy #{req.method} #{req.url}"

    self = @

    if isFunction args[0]
      callback = args[0]
      metaVariables = args[0]
      callback = args[1]

Pause request so we don't miss any data or end events.

    resume = pause req

    @createConnection (err, connection) ->
      if err
        if callback then callback err
        else self.emit 'error', err
        if callback
          connection.on 'close', callback
          connection.on 'error', (error) ->
            connection.removeListener 'close', callback
            callback error

        connection.proxyRequest req, res, metaVariables

Flush any events captured while we were establishing our client connection


Send SIGKILL to process. This will kill it for sure.

  kill: ->
    debug "process kill"

    if @child
      @changeState 'quitting'
      @child.kill 'SIGKILL'
      @heartbeat.destroy() if @heartbeat

Send SIGTERM to process. This will immediately kill it.

  terminate: ->
    debug "process terminate"

    if @child
      @changeState 'quitting'
      @child.kill 'SIGTERM'
      @heartbeat.destroy() if @heartbeat

      timeout = setTimeout =>
        if @state is 'quitting'
      , 10000
      @once 'exit', -> clearTimeout timeout

Send SIGTERM to process. The process will finish serving its request and gracefully quit.

  quit: ->
    debug "process quit"

    if @child
      @changeState 'quitting'
      @child.kill 'SIGQUIT'
      @heartbeat.destroy() if @heartbeat

      timeout = setTimeout =>
        if @state is 'quitting'
      , 3000
      @once 'exit', -> clearTimeout timeout

Quit and respawn process

  restart: ->
    debug "process restart"

    @once 'exit', => @spawn()

Public API for creating a Process

exports.createProcess = (args...) ->
  new Process args...

Generates a random path.

tmpFile = ->
  pid  = process.pid
  rand = Math.floor Math.random() * 10000000000
  "/tmp/nack." + pid + "." + rand

TODO: Don't poll FS

onceFileExists = (path, callback, timeout = 3000) ->
  timeoutError = null
  timeoutId = setTimeout ->
    timeoutError = new Error "timeout: waiting for #{path}"
  , timeout

  decay = 1

  statPath = (err, stat) ->
    if !err and stat and stat.isSocket()
      clearTimeout timeoutId
      callback err, path
    else if timeoutError
      callback timeoutError, path
      setTimeout ->
        fs.stat path, statPath
      , decay *= 2


tryConnect = (connection, path, callback) ->
  errors = 0

  reconnect = ->
    onceFileExists path, (err) ->
      return callback err if err
      connection.connect path

  onError = (err) ->
    if err and ++errors > 3
      connection.removeListener 'error', onError
      callback new Error "timeout: couldn't connect to #{path}"

  connection.on 'error', onError

  connection.on 'connect', ->
    connection.removeListener 'error', onError
    callback null, connection