Initial profiles support. Needs more testing.

acs.coffee

@@ -6,18 +6,41 @@ tr069 = require './tr-069'
 tasks = require './tasks'
 sanitize = require('./sanitize').sanitize
 db = require './db'
+profiles = require './profiles'
 
 
-currentClientIP = null
+currentRequest = null
 
-writeResponse = (serverResponse, res) ->
+
+applyConfigurations = (configurations) ->
+  taskList = []
+  for c of configurations
+    task = {device : currentRequest.deviceId, name : 'setParameterValues', parameterValues: [], timestamp : db.mongo.Timestamp()}
+    task.parameterValues.push([c, configurations[c]])
+    taskList.push task
+
+  if taskList.length
+    util.log("#{currentRequest.deviceId}: Profiles discrepancy found")
+    db.tasksCollection.save(taskList, (err) ->
+      for task in taskList
+        util.log("#{currentRequest.deviceId}: Added #{task.name} task #{task._id}")
+      task = taskList[0]
+      util.log("#{currentRequest.deviceId}: started task #{task.name}(#{task._id})")
+      runTask(task, {})
+    )
+  else
+    res = tr069.response(currentRequest.sessionId, {}, {})
+    writeResponse res
+
+
+writeResponse = (res) ->
   if config.DEBUG
     s = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n"
-    db.memcached.append("debug-#{currentClientIP}", s, (err, result) ->
+    db.memcached.append("debug-#{currentRequest.clientIp}", s, (err, result) ->
     )
 
-  serverResponse.writeHead(res.code, res.headers)
-  serverResponse.end(res.data)
+  currentRequest.response.writeHead(res.code, res.headers)
+  currentRequest.response.end(res.data)
 
 
 updateDevice = (deviceId, actions, callback) ->
@@ -50,21 +73,21 @@ updateDevice = (deviceId, actions, callback) ->
       updates["#{path}_timestamp"] = now
 
   if Object.keys(updates).length > 0
-    db.devicesCollection.update({'_id' : deviceId}, {'$set' : updates}, {safe: true}, (err, count) ->
+    db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {safe: true}, (err, count) ->
       if (err)
         callback(err) if callback?
         return
 
       if count == 0
-        util.log("#{deviceId}: new device detected")
-        db.devicesCollection.update({'_id' : deviceId}, {'$set' : updates}, {upsert: true, safe: true}, (err) ->
+        util.log("#{currentRequest.deviceId}: new device detected")
+        db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {upsert: true, safe: true}, (err) ->
           if err?
             callback(err) if callback?
             return
           
-          task = {device : deviceId, name : 'init', timestamp : db.mongo.Timestamp()}
+          task = {device : currentRequest.deviceId, name : 'init', timestamp : db.mongo.Timestamp()}
           db.tasksCollection.save(task, (err) ->
-            util.log("#{deviceId}: Added init task #{task._id}")
+            util.log("#{currentRequest.deviceId}: Added init task #{task._id}")
             callback(err) if callback?
           )
         )
@@ -75,39 +98,63 @@ updateDevice = (deviceId, actions, callback) ->
     callback() if callback?
 
 
-runTask = (sessionId, deviceId, task, reqParams, response) ->
+runTask = (task, reqParams) ->
   resParams = {}
   actions = tasks.task(task, reqParams, resParams)
-  updateDevice(deviceId, actions)
+  updateDevice(currentRequest.deviceId, actions)
 
   if Object.keys(resParams).length > 0
     db.memcached.set(String(task._id), task, config.CACHE_DURATION, (err, result) ->
-      res = tr069.response(sessionId, resParams, {task : String(task._id)})
-      writeResponse response, res
+      res = tr069.response(currentRequest.sessionId, resParams, {task : String(task._id)})
+      writeResponse res
     )
     return
 
   # Task finished
   db.memcached.del(String(task._id))
   db.tasksCollection.remove({'_id' : db.mongo.ObjectID(task._id)}, {safe: true}, (err, removed) ->
-    util.log("#{deviceId}: completed task #{task.name}(#{task._id})")
+    util.log("#{currentRequest.deviceId}: completed task #{task.name}(#{task._id})")
     cookies = {task : null}
-    nextTask(sessionId, deviceId, response, cookies)
+    nextTask(cookies)
   )
 
 
-nextTask = (sessionId, deviceId, response, cookies) ->
-  cur = db.tasksCollection.find({'device' : deviceId}).sort(['timestamp']).limit(1)
+nextTask = (cookies) ->
+  cur = db.tasksCollection.find({'device' : currentRequest.deviceId}).sort(['timestamp']).limit(1)
   cur.nextObject( (err, task) ->
     reqParams = {}
     resParams = {}
+
     if not task
-      res = tr069.response(sessionId, resParams, cookies)
-      writeResponse response, res
+      # no more taks, check profile discrepancy
+      db.memcached.get(["#{currentRequest.deviceId}_profiles_hash", 'profiles_hash'], (err, results) ->
+        profilesHash = results['profiles_hash']
+        deviceProfilesHash = results["#{currentRequest.deviceId}_profiles_hash"]
+
+        if not deviceProfilesHash or profilesHash != deviceProfilesHash
+          profiles.findDeviceConfigurationDiscrepancy(currentRequest.deviceId, profilesHash, (configurations) ->
+            applyConfigurations(configurations)
+          )
+        else if not profilesHash
+          profiles.getProfilesHash((hash) ->
+            if hash != deviceProfilesHash
+              profiles.findDeviceConfigurationDiscrepancy(currentRequest.deviceId, profilesHash, (configurations) ->
+                applyConfigurations(configurations)
+              )
+            else
+              # no discrepancy, return empty response
+              res = tr069.response(currentRequest.sessionId, resParams, cookies)
+              writeResponse res
+          )
+        else
+          # no discrepancy, return empty response
+          res = tr069.response(currentRequest.sessionId, resParams, cookies)
+          writeResponse res
+      )
       return
 
-    util.log("#{deviceId}: started task #{task.name}(#{task._id})")
-    runTask(sessionId, deviceId, task, reqParams, response)
+    util.log("#{currentRequest.deviceId}: started task #{task.name}(#{task._id})")
+    runTask(task, reqParams)
   )
 
 
@@ -131,12 +178,13 @@ if cluster.isMaster
 else
   if config.DEBUG
     process.on('uncaughtException', (error) ->
+      util.error(error.stack)
       # dump request/response logs and stack trace
-      util.log("Unexpected error occured. Writing log to debug/#{currentClientIP}.log.")
-      db.memcached.get("debug-#{currentClientIP}", (err, l) ->
+      db.memcached.get("debug-#{currentRequest.clientIp}", (err, l) ->
+        util.log("Unexpected error occured. Writing log to debug/#{currentRequest.clientIp}.log.")
         util.error(err) if err
         fs = require 'fs'
-        fs.writeFileSync("debug/#{currentClientIP}.log", l + "\n\n" + error.stack)
+        fs.writeFileSync("debug/#{currentRequest.clientIp}.log", l + "\n\n" + error.stack)
         process.exit(1)
       )
     )
@@ -169,12 +217,15 @@ else
       return body.toString(encoding || 'utf8', 0, body.byteLength)
 
     request.addListener 'end', () ->
-      currentClientIP = request.connection.remoteAddress
+      currentRequest = {}
+      currentRequest.request = request
+      currentRequest.response = response
+      currentRequest.clientIp = request.connection.remoteAddress
       if config.DEBUG
         s = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(request.headers) + "\n#{request.getBody()}\n\n"
-        db.memcached.append("debug-#{currentClientIP}", s, (err, result) ->
+        db.memcached.append("debug-#{currentRequest.clientIp}", s, (err, result) ->
           if err == 'Item is not stored'
-            db.memcached.set("debug-#{currentClientIP}", s, config.CACHE_DURATION, (err, result) ->
+            db.memcached.set("debug-#{currentRequest.clientIp}", s, config.CACHE_DURATION, (err, result) ->
             )
         )
 
@@ -183,26 +234,27 @@ else
 
       if reqParams.inform
         resParams.inform = true
-        cookies.ID = sessionId = reqParams.sessionId
-        cookies.DeviceId = deviceId = common.getDeviceId(reqParams.deviceId)
+        cookies.ID = currentRequest.sessionId = reqParams.sessionId
+        cookies.DeviceId = currentRequest.deviceId = common.getDeviceId(reqParams.deviceId)
         if config.DEBUG
-          util.log("#{deviceId}: inform (#{reqParams.eventCodes}); retry count #{reqParams.retryCount}")
+          util.log("#{currentRequest.deviceId}: inform (#{reqParams.eventCodes}); retry count #{reqParams.retryCount}")
 
-        updateDevice(deviceId, {'inform' : true, 'parameterValues' : reqParams.informParameterValues}, (err) ->
-          res = tr069.response(sessionId, resParams, cookies)
-          writeResponse response, res
+        updateDevice(currentRequest.deviceId, {'inform' : true, 'parameterValues' : reqParams.informParameterValues}, (err) ->
+          res = tr069.response(currentRequest.sessionId, resParams, cookies)
+          writeResponse res
         )
         return
 
-      sessionId = reqParams.cookies.ID
-      deviceId = reqParams.cookies.DeviceId
+      currentRequest.sessionId = reqParams.cookies.ID
+      currentRequest.deviceId = reqParams.cookies.DeviceId
       taskId = reqParams.cookies.task
 
       if not taskId
-        nextTask(sessionId, deviceId, response, cookies)
+        nextTask(cookies)
       else
         db.memcached.get(taskId, (err, task) ->
-          runTask(sessionId, deviceId, task, reqParams, response)
+          # TODO if not found in memcached, fetch from mongo
+          runTask(task, reqParams)
         )
 
   server.listen config.ACS_PORT, config.ACS_INTERFACE

api.coffee

@@ -107,10 +107,28 @@ else
       body = request.getBody()
       urlParts = url.parse(request.url, true)
       if PROFILES_REGEX.test(urlParts.pathname)
+        profileName = PROFILES_REGEX.exec(urlParts.pathname)[1]
         if request.method == 'PUT'
-          # TODO save profile
+          profile = JSON.parse(body)
+          profile._id = profileName
+
+          profilesCollection.save(profile, (err) ->
+            if err
+              response.writeHead(500)
+              response.end(err)
+              return
+            response.writeHead(200)
+            response.end()
+          )
         else if request.method == 'DELETE'
-          # TODO delete profile
+          profilesCollection.remove({'_id' : profileName}, (err, removedCount) ->
+            if err
+              response.writeHead(500)
+              response.end(err)
+              return
+            response.writeHead(200)
+            response.end()
+          )
         else
           response.writeHead 405, {'Allow': 'PUT, DELETE'}
           response.end('405 Method Not Allowed')

config.coffee

@@ -3,7 +3,8 @@ exports.ACS_PORT = 7547 # same as AP Manager's port
 exports.ACS_INTERFACE = '0.0.0.0'
 exports.API_PORT = 7557
 exports.API_INTERFACE = '0.0.0.0'
-exports.CACHE_DURATION = 60000 # in milliseconds
+exports.CACHE_DURATION = 60 # in seconds
+exports.PROFILES_CACHE_DURATION = 86400
 exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock'
 exports.MEMCACHED_SOCKET = '/tmp/memcached.sock'
 exports.WORKER_RESPAWN_TIME = 3000

db.coffee

@@ -1,13 +1,15 @@
 config = require './config'
 mongo = require 'mongodb'
 Memcached = require 'memcached'
-
-exports.memcached = new Memcached(config.MEMCACHED_SOCKET)
+Memcached.config.maxExpiration = 86400
+memcached = new Memcached(config.MEMCACHED_SOCKET)
 
 # Create MongoDB connections
-dbserver = new mongo.Server(config.MONGODB_SOCKET, 0, {auto_reconnect: true})
 tasksCollection = null
 devicesCollection = null
+profilesCollection = null
+
+dbserver = new mongo.Server(config.MONGODB_SOCKET, 0, {auto_reconnect: true})
 db = new mongo.Db(config.DATABASE_NAME, dbserver, {native_parser:true, safe:true})
 
 db.open( (err, db) ->
@@ -20,6 +22,11 @@ db.open( (err, db) ->
   db.collection('devices', (err, collection) ->
     exports.devicesCollection = collection
   )
+
+  db.collection('profiles', (err, collection) ->
+    exports.profilesCollection = collection
+  )
 )
 
-exports.mongo = mongo
+exports.mongo = mongo
+exports.memcached = memcached

profiles.coffee

@@ -0,0 +1,123 @@
+config = require './config'
+db = require './db'
+
+
+getProfiles = (callback) ->
+  db.memcached.get('profiles', (err, res) ->
+    throw err if err
+
+    if res
+      callback(res)
+      return
+
+    db.profilesCollection.find().toArray((err, res) ->
+      if err
+        callback(err, res)
+        return
+
+      profiles = {}
+      for p in res
+        profiles[p['_id']] = p
+
+      db.memcached.set('profiles', profiles, config.PROFILES_CACHE_DURATION)
+      callback(profiles)
+    )
+  )
+
+
+getProfilesHash = (callback) ->
+  db.memcached('profiles_hash', (err, res) ->
+    if res
+      callback(res)
+    else
+      getProfiles((profiles) ->
+        hash = callback(calculateProfilesHash(profiles))
+        db.memcached.set('profiles_hash', hash, config.PROFILES_CACHE_DURATION, (err, res) ->
+          callback(hash)
+        )
+      )
+  )
+
+
+calculateProfilesHash = (profiles) ->
+  crypto = require('crypto')
+  hash = crypto.createHash('md5').update(JSON.stringify(profiles)).digest('hex')
+  return hash
+
+
+getProfilesVaraibles = (profiles) ->
+  vars = []
+  for p in profiles
+    for c in p.configurations
+      vars.push(c[0])
+  return vars
+
+
+exports.findDeviceConfigurationDiscrepancy = (deviceId, profilesHash, callback) ->
+  getProfiles((profiles) ->
+    params = {'_tags' : 1}
+    for p of profiles
+      profile = profiles[p]
+      for c in profile.configurations
+        switch c.type
+          when 'value'
+            params[c.param] = 1
+          else
+            throw 'Unknown configuration type'
+
+    db.devicesCollection.findOne({'_id' : deviceId}, params, (err, device) ->
+      tags = device['_tags'] or []
+      tags.push('default')
+
+      deviceProfiles = []
+      for t in tags
+        if profiles[t]?
+          deviceProfiles.push(profiles[t])
+
+      combinedConfigurations = combineProfileConfigurations(deviceProfiles)
+
+      discrepency = {}
+      for c in combinedConfigurations
+        switch c.type
+          when 'value'
+            if c.value != getDeviceParamFromPath(device, "#{c.param}._value")
+              discrepency[c.param] = c.value
+          else
+            throw 'Unknown configuration type'
+      if not profilesHash
+        profilesHash = calculateProfilesHash(profiles)
+        db.memcached.set('profiles_hash', profilesHash, config.PROFILES_CACHE_DURATION, (err, res) ->
+        )
+
+      db.memcached.set("#{deviceId}_profiles_hash", profilesHash, config.PROFILES_CACHE_DURATION, (err, res) ->
+        callback(discrepency)
+      )
+    )
+  )
+
+
+combineProfileConfigurations = (profiles) ->
+  maxWeights = {}
+  configurations = {}
+  for p in profiles
+    for c in p.configurations
+      type = c.type
+      switch type
+        when 'value'
+          paramName = c.param
+          if not maxWeights["#{type}_#{paramName}"]? or p.weight > maxWeights["#{type}_#{paramName}"]
+            configurations["#{type}_#{paramName}"] = c
+            maxWeights["#{type}_#{paramName}"] = p.weight
+        else
+          throw 'Unknown configuration type'
+
+  configurationsList = (configurations[c] for c of configurations)
+  return configurationsList
+
+
+getDeviceParamFromPath = (device, paramPath) ->
+  path = paramPath.split('.')
+  ref = device
+  for p in path
+    ref = ref[p]
+  return ref