genieacs/acs.coffee

config = require './config'
common = require './common'
util = require 'util'
http = require 'http'
mongo = require 'mongodb'
tr069 = require './tr-069'
tasks = require './tasks'
Memcached = require 'memcached'
memcached = new Memcached(config.MEMCACHED_SOCKET)
sanitize = require('./sanitize').sanitize


# Create MongoDB connections
dbserver = new mongo.Server(config.MONGODB_SOCKET, 0, {auto_reconnect: true})
tasksCollection = null
devicesCollection = null
db = new mongo.Db(config.DATABASE_NAME, dbserver, {native_parser:true})
db.open( (err, db) ->
  db.collection('tasks', (err, collection) ->
    tasksCollection = collection
    tasksCollection.ensureIndex({device: 1, timestamp: 1})
  )

  db.collection('devices', (err, collection) ->
    devicesCollection = collection
  )
)


updateDevice = (deviceId, actions, callback) ->
  if not actions?
    callback() if callback?
    return

  updates = {}
  if actions.parameterValues?
    for p in actions.parameterValues
      v = sanitize(p[0], p[1])
      path = if common.endsWith(p[0], '.') then p[0] else "#{p[0]}."
      if v is undefined
        updates["#{path}_value"] = p[1]
      else
        updates["#{path}_value"] = v
        updates["#{path}_orig"] = p[1]

  if actions.parameterNames?
    for p in actions.parameterNames
      if common.endsWith(p[0], '.')
        updates["#{p[0]}_writable"] = p[1]
        updates["#{p[0]}_object"] = true
      else
        updates["#{p[0]}._writable"] = p[1]

  if Object.keys(updates).length > 0
    devicesCollection.update({'_id' : deviceId}, {'$set' : updates}, {upsert: true, safe: true}, (err) ->
      callback(err) if callback?
    )
  else
    callback() if callback?


runTask = (sessionId, deviceId, task, reqParams, response) ->
  resParams = {}
  actions = tasks.task(task, reqParams, resParams)
  updateDevice(deviceId, actions)

  if Object.keys(resParams).length > 0
    memcached.set(String(task._id), task, config.CACHE_DURATION, (err, result) ->
      tr069.response(reqParams.sessionId, response, resParams, {task : String(task._id)})
    )
    return

  # Task finished
  memcached.del(String(task._id))
  tasksCollection.remove({'_id' : mongo.ObjectID(task._id)}, {safe: true}, (err, removed) ->
    util.log("#{deviceId}: completed task #{task.name}(#{task._id})")
    cookies = {task : null}
    tr069.response(sessionId, response, resParams, cookies)
    nextTask(sessionId, deviceId, response, cookies)
  )


nextTask = (sessionId, deviceId, response, cookies) ->
  cur = tasksCollection.find({'device' : deviceId}).sort(['timestamp']).limit(1)
  cur.nextObject( (err, task) ->
    reqParams = {}
    resParams = {}
    if not task
      tr069.response(sessionId, response, resParams, cookies)
      return

    util.log("#{deviceId}: started task #{task.name}(#{task._id})")
    runTask(sessionId, deviceId, task, reqParams, response)
  )


cluster = require 'cluster'
numCPUs = require('os').cpus().length

if cluster.isMaster
  cluster.on('listening', (worker, address) ->
    util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
  )

  cluster.on('exit', (worker, code, signal) ->
    util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode})")
    setTimeout(()->
      cluster.fork()
    , config.WORKER_RESPAWN_TIME)
  )

  for i in [1 .. numCPUs]
    cluster.fork()
else
  server = http.createServer (request, response) ->
    if request.method != 'POST'
      #console.log '>>> 405 Method Not Allowed'
      response.writeHead 405, {'Allow': 'POST'}
      response.end('405 Method Not Allowed')
      return

    request.content = ''
    cookies = {}

    request.addListener 'data', (chunk) ->
      request.content += chunk

    request.addListener 'end', () ->
      resParams = {}
      reqParams = tr069.request(request.headers, request.content)

      if reqParams.inform
        resParams.inform = true
        cookies.ID = sessionId = reqParams.sessionId
        cookies.DeviceId = deviceId = common.getDeviceId(reqParams.deviceId)
        util.log("#{deviceId}: inform (#{reqParams.eventCodes}); retry count #{reqParams.retryCount}")
        devicesCollection.count({'_id' : deviceId}, (err, count) ->
          if not count
            util.log("#{deviceId}: new device detected")
            task = {device : deviceId, name : 'init', timestamp : mongo.Timestamp()}
            tasksCollection.save(task, (err) ->
              util.log("#{deviceId}: Added init task #{task._id}")
            )

          updateDevice(deviceId, {'parameterValues' : reqParams.informParameterValues}, (err) ->
            tr069.response(reqParams.sessionId, response, resParams, cookies)
          )
        )
        return

      sessionId = reqParams.cookies.ID
      deviceId = reqParams.cookies.DeviceId
      taskId = reqParams.cookies.task

      if not taskId
        nextTask(sessionId, deviceId, response, cookies)
        return
      else
        memcached.get(taskId, (err, task) ->
          runTask(sessionId, deviceId, task, reqParams, response)
        )

  server.listen config.ACS_PORT, config.ACS_INTERFACE
  #console.log "Server listening on port #{config.ACS_PORT}"