amqp = require 'amqp'

hat = require 'hat'

os = require 'os'

{EventEmitter} = require 'events'

fs = require "fs"

nodePath = require 'path'

{environment} = require 'koding-config-manager'

class Worker extends EventEmitter

@throwIt = (it) ->

console.error it if it?

constructor:(options = {},data)->

options.name or= "worker"

options.serviceGenericName or= ""

options.serviceUniqueName or= ""

options.status or= 1

options.command or= ""

options.option or= "many"

options.uuid or= hat()

options.hostname or= os.hostname()

options.pid or= process.pid

options.proxyName or= ""

options.interval or= 10

options.port or= 0

options.rabbitKey or= ""

options.binary or= no

options.mq or=

host : 'localhost'

port : '5672'

login : 'guest'

password : 'guest'

vhost : '/'

super options, data

@defaultInit options, data

defaultInit:(options, data)->

@reporter = {}

@interval = options.interval

@amqpOptions = options.mq

@binary = options.binary

@worker =

name : options.name

proxyName : options.proxyName

serviceGenericName : options.serviceGenericName

serviceUniqueName : options.serviceUniqueName

message :

command : options.command

option : options.option

uuid : options.uuid

hostname : options.hostname

timestamp : @isoDateString()

pid : options.pid

status : options.status

version : parseInt(@getVersion())

environment : environment

port : options.port

monitor :

mem :

rss : 0

heapTotal : 0

heapUsed : 0

unit : 'MB'

uptime : 0

publish: (command) ->

@fetchWorkerExchange (workerExchange) =>

@worker.message.command = command

@worker.timestamp = @isoDateString()

@worker.monitor.uptime += @interval

workerExchange.publish "input.worker", { @worker },

appId: "#{@worker.uuid}"

startReporter: (command) ->

@reporter[command] = setInterval =>

@publish command

, @interval * 1000

stopReporter: (command) ->

clearInterval @reporter[command]

isoDateString: ->

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"

getVersion:() ->

versionFile = nodePath.join(__dirname, '../../VERSION')

if fs.existsSync versionFile

version = (fs.readFileSync versionFile, 'utf-8').trim()

return version ? "0.0.1"

getRabbitKey:() ->

version = (fs.readFileSync nodePath.join(__dirname, '../../KEY'), 'utf-8').trim()

handleMessage:(message, headers, deliveryInfo)=>

messageData = JSON.parse message.data.toString("utf-8")

unless messageData.name?

return

unless messageData.uuid is @worker.uuid

return

switch messageData.command

when 'start', 'add', 'first.start' #backward compability, last two keywords will be removed

@emit 'permissionToRun'

when 'noPermission', 'added.before' #same as above, backward compability

console.log "[#{@worker.name} - #{@isoDateString()}] no permission. #{messageData.log}"

when 'kill'

@emit 'killWorkerProcess'

when 'killForce'

@emit 'killForceWorkerProcess'

else

console.log "received command unknown: #{messageData.command}"

fetchExchange:(name, options, callback)->

exchangeName = "#{name}Exchange"

readyEvent = "#{exchangeName}Ready"

if @[exchangeName] is null

@once readyEvent, => @fetchWorkerExchange callback

else unless @workerExchange?

@[exchangeName] = null

@connection.exchange exchangeName, options,

(exchange)=>

@[exchangeName] = exchange

@emit readyEvent

callback exchange

else callback @[exchangeName]

fetchWorkerExchange:(callback)->

@fetchExchange 'worker', @getWorkerExchangeOptions(), callback

getWorkerExchangeOptions:->

type : 'topic'

autoDelete : no

durable : true

connect: ()->

{host, port, protocol, login, password, vhost, heartbeat, autoReconnect} = @amqpOptions

port ?= 5672

protocol ?= 'amqp:'

options = {host, port, protocol} # but not vhost, because of a bug in node-amqp

options.login = login if login?

options.password = password if password?

options.vhost = vhost if vhost?

options.heartbeat = heartbeat if heartbeat?

options.autoReconnect = autoReconnect ? no

@connection = amqp.createConnection options

@connection.on "error", Worker.throwIt

@connection.on "ready", =>

@connection.exchange "workerExchange", @getWorkerExchangeOptions(), (exchange) =>

@connection.queue "#{@worker.name}-#{@worker.hostname}-#{@worker.uuid}",

durable: false

autoDelete: true

, (queue) =>

queue.bind 'workerExchange', "output.worker.#{@worker.uuid}"

queue.on "queueBindOk", =>

unless @binary

if @worker.message.option is "one" or @worker.message.option is "version"

@startReporter "add"

else

@fetchWorkerExchange (workerExchange)=>

workerExchange.publish "input.worker", {worker: @worker},

appId: "#{@worker.uuid}"

else

@emit 'permissionToRun'

queue.subscribe (message, headers, deliveryInfo) =>

@handleMessage message, headers, deliveryInfo

module.exports = Worker