From 3348139795ae8900cf92ca2c29a4dc05aa97f2fe Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Mon, 7 Jul 2014 23:32:42 +0300 Subject: [PATCH] Moved detection of removed parameters to updateDevice function plus other organization and optimization changes --- lib/cwmp.coffee | 92 +------------------------ lib/mongo-query.coffee | 1 + lib/tasks.coffee | 59 ++-------------- lib/update-device.coffee | 143 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 152 insertions(+), 143 deletions(-) create mode 100644 lib/update-device.coffee diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index 1b43b34..2bbd145 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -5,7 +5,6 @@ http = require 'http' https = require 'https' soap = require './soap' tasks = require './tasks' -normalize = require('./normalize').normalize db = require './db' presets = require './presets' mongodb = require 'mongodb' @@ -13,11 +12,8 @@ fs = require 'fs' apiFunctions = require './api-functions' customCommands = require './custom-commands' crypto = require 'crypto' -parameters = require './parameters' zlib = require 'zlib' - -OBJECT_REGEX = /\.$/ -INSTANCE_REGEX = /\.[\d]+\.$/ +updateDevice = require './update-device' # Used to reject new TR-069 sessions when under load @@ -61,88 +57,6 @@ endSession = (currentRequest) -> ) -updateDevice = (currentRequest, actions, callback) -> - if not actions? - callback() if callback? - return - - now = new Date(Date.now()) - update = {'$set' : {}, '$unset' : {}} - - if actions.parameterValues? - for p in actions.parameterValues - origValue = p[1] - v = normalize(p[0], origValue) - - path = if common.endsWith(p[0], '.') then p[0] else "#{p[0]}." - if v == origValue - update['$unset']["#{path}_orig"] = 1 - else - update['$set']["#{path}_orig"] = origValue - update['$set']["#{path}_value"] = v - update['$set']["#{path}_timestamp"] = now - if p[2]? - update['$set']["#{path}_type"] = p[2] - else - update['$unset']["#{path}_type"] = 1 - - if actions.deletedObjects? - for p in actions.deletedObjects - update['$unset'][p] = 1 - - if actions.instanceName? - for i in actions.instanceName - update['$set']["#{i[0]}._name"] = i[1] - - if actions.parameterNames? - for p in actions.parameterNames - if OBJECT_REGEX.test(p[0]) - path = p[0] - if INSTANCE_REGEX.test(p[0]) - update['$set']["#{path}_instance"] = true - else - update['$set']["#{path}_object"] = true - else - path = "#{p[0]}." - update['$set']["#{path}_writable"] = p[1] if p[1]? - update['$set']["#{path}_timestamp"] = now - - # Add parameter (sans any dot suffix) if has an alias - if (a = parameters.getAlias(pp = path.slice(0, -1)))? - newAliases = {} if not newAliases? - newAliases[pp] = a - - if actions.customCommands? - for p in actions.customCommands - commandName = p[0] - update['$set']["_customCommands.#{commandName}._value"] = p[1] - update['$set']["_customCommands.#{commandName}._timestamp"] = now - - if actions.set? - common.extend(update['$set'], actions.set) - - for k of update - delete update[k] if Object.keys(update[k]).length == 0 - - if Object.keys(update).length - db.devicesCollection.update({'_id' : currentRequest.session.deviceId}, update, {}, (err, count) -> - throw err if err - # Clear aliases cache if there's a new aliased parameter that has not been included in aliases cache - if newAliases? - db.getAliases((aliases) -> - for p, a of newAliases - if not aliases[a]? or p not in aliases[a] - db.redisClient.del('aliases', (err, res) -> - throw err if err - ) - break - ) - callback?(err) - ) - else - callback?() - - inform = (currentRequest, cwmpRequest) -> # If overloaded, ask CPE to retry in 5 mins if Date.now() < holdUntil @@ -172,7 +86,7 @@ inform = (currentRequest, cwmpRequest) -> throw err if err updateAndRespond = () -> - updateDevice(currentRequest, actions, (err) -> + updateDevice(currentRequest.session.deviceId, actions, (err) -> throw err if err res = soap.response({ id : cwmpRequest.id, @@ -273,7 +187,7 @@ runTask = (currentRequest, task, methodResponse) -> tasks.task(task, methodResponse, (err, status, methodRequest, deviceUpdates) -> throw err if err # TODO handle error - updateDevice(currentRequest, deviceUpdates, (err) -> + updateDevice(currentRequest.session.deviceId, deviceUpdates, (err) -> throw err if err timeDiff = process.hrtime(timeDiff)[0] + 1 diff --git a/lib/mongo-query.coffee b/lib/mongo-query.coffee index 61b67d0..73f5369 100644 --- a/lib/mongo-query.coffee +++ b/lib/mongo-query.coffee @@ -93,6 +93,7 @@ optimizeProjection = (obj) -> delete obj[b] keys.splice(i--, 1) ++ i + return exports.test = test diff --git a/lib/tasks.coffee b/lib/tasks.coffee index 9aef892..2faf8b8 100644 --- a/lib/tasks.coffee +++ b/lib/tasks.coffee @@ -53,31 +53,6 @@ this.refreshObject = (task, methodResponse, callback) -> throw Error('Unexpected subtask name') -findMissingParameters = (device, parameterList, prefix) -> - missingParameters = [] - paths = {} - for param in parameterList - p = param[0] - paths[p] = true - i = p.indexOf('.') - while i != -1 - paths[p[0...i]] = true - i = p.indexOf('.', i + 1) - - recursive = (obj, prefix) -> - for k,v of obj - continue if k[0] == '_' - p = prefix + k - - if paths[p] - recursive(v, "#{p}.") - else - missingParameters.push(p) - - recursive(device, prefix) - return missingParameters - - this.getParameterNames = (task, methodResponse, callback) -> if methodResponse.faultcode? task.fault = methodResponse @@ -99,42 +74,18 @@ this.getParameterNames = (task, methodResponse, callback) -> for k, v of found methodResponse.parameterList.push([k]) if v == 0 + # Some devices don't return the root object as described in the standard. add manually to update timestamp + methodResponse.parameterList.push([task.parameterPath]) if !!task.parameterPath + deviceUpdates = {parameterNames : methodResponse.parameterList} - path = if common.endsWith(task.parameterPath, '.') then task.parameterPath.slice(0, -1) else task.parameterPath - projection = {} - projection[path] = 1 if !!task.parameterPath - - # delete nonexisting params - db.devicesCollection.findOne({_id : task.device}, projection, (err, device) -> - if device - root = device - rootPath = '' - if !!task.parameterPath - ps = path.split('.') - for p in ps - rootPath += "#{p}." - root = root[p] - break if not root? - - if root - deviceUpdates.deletedObjects = findMissingParameters(root, methodResponse.parameterList, rootPath) - else - # avoid adding and deleting the same param - if rootPath isnt task.parameterPath - deviceUpdates.deletedObjects = [rootPath.slice(0, -1)] - - # some devices don't return the root object as described in the standard. add manually to update timestamp - deviceUpdates.parameterNames.push([task.parameterPath]) if !!task.parameterPath - - callback(null, STATUS_COMPLETED, null, deviceUpdates) - ) + return callback(null, STATUS_COMPLETED, null, deviceUpdates) else methodRequest = { type : 'GetParameterNames', parameterPath : task.parameterPath, nextLevel : if task.nextLevel? then task.nextLevel else false } - callback(null, STATUS_OK, methodRequest) + return callback(null, STATUS_OK, methodRequest) this.getParameterValues = (task, methodResponse, callback) -> diff --git a/lib/update-device.coffee b/lib/update-device.coffee new file mode 100644 index 0000000..c587dbb --- /dev/null +++ b/lib/update-device.coffee @@ -0,0 +1,143 @@ +common = require './common' +db = require './db' +normalize = require('./normalize').normalize +mongoQuery = require './mongo-query' +parameters = require './parameters' + +OBJECT_REGEX = /\.$/ +INSTANCE_REGEX = /\.[\d]+\.$/ + + +updateDevice = (deviceId, actions, callback) -> + return callback?() if not actions? + + now = new Date(Date.now()) + update = {'$set' : {}, '$unset' : {}} + + if actions.parameterValues? + for p in actions.parameterValues + origValue = p[1] + v = normalize(p[0], origValue) + + path = if common.endsWith(p[0], '.') then p[0] else "#{p[0]}." + if v == origValue + update['$unset']["#{path}_orig"] = 1 + else + update['$set']["#{path}_orig"] = origValue + update['$set']["#{path}_value"] = v + update['$set']["#{path}_timestamp"] = now + if p[2]? + update['$set']["#{path}_type"] = p[2] + + if actions.deletedObjects? + for p in actions.deletedObjects + update['$unset'][p] = 1 + + if actions.instanceName? + for i in actions.instanceName + update['$set']["#{i[0]}._name"] = i[1] + + if actions.parameterNames? + projection = {} # fetch only necessary parameters + presentPaths = {} # to be compared with to find no longer existing parameters + + for p in actions.parameterNames + if OBJECT_REGEX.test(p[0]) + path = p[0].slice(0, -1) + if INSTANCE_REGEX.test(p[0]) + update['$set']["#{path}._instance"] = true + else + update['$set']["#{path}._object"] = true + else + path = p[0] + update['$set']["#{path}._writable"] = p[1] if p[1]? + update['$set']["#{path}._timestamp"] = now + + # Add parameter (sans any dot suffix) if has an alias + if (a = parameters.getAlias(path))? + newAliases = {} if not newAliases? + newAliases[path] = a + + projection[path] = 1 + + # Store this param path and all its parents for later look up + presentPaths[path] = true + i = p[0].indexOf('.') + while i != -1 + presentPaths[p[0][0...i]] = true + i = p[0].indexOf('.', i + 1) + + mongoQuery.optimizeProjection(projection) + + if actions.customCommands? + for p in actions.customCommands + commandName = p[0] + update['$set']["_customCommands.#{commandName}._value"] = p[1] + update['$set']["_customCommands.#{commandName}._timestamp"] = now + + if actions.set? + common.extend(update['$set'], actions.set) + + f = () -> + for k of update + delete update[k] if Object.keys(update[k]).length == 0 + + if Object.keys(update).length + db.devicesCollection.update({'_id' : deviceId}, update, {}, (err, count) -> + throw err if err + # Clear aliases cache if there's a new aliased parameter that has not been included in aliases cache + if newAliases? + db.getAliases((aliases) -> + for p, a of newAliases + if not aliases[a]? or p not in aliases[a] + db.redisClient.del('aliases', (err, res) -> + throw err if err + ) + break + ) + callback?(err) + ) + else + callback?() + + if projection? + db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> + throw err if err + + recursive = (obj, prefix) -> + for k,v of obj + p = prefix + k + if k[0] == '_' + # Remove set operations for unmodified values for optimization + if update['$set'][p] == v + delete update['$set'][p] + continue + + # Remove unset operations for values that don't exist + delete update['$unset']["#{p}._orig"] if not v._orig? + + if presentPaths[p] + recursive(v, "#{p}.") + else + update['$unset'][p] = 1 + + # Scan the DB object and remove parameters that no longer exists in device + for k, v of projection + root = device + rootPath = '' + for p in path.split('.') + rootPath += "#{p}." + root = root[p] + break if not root? + + if root + recursive(root, rootPath) + else + if not presentPaths[rootPath.slice(0, -1)] + update['$unset'][rootPath.slice(0, -1)] = 1 + f() + ) + else + f() + +module.exports = updateDevice \ No newline at end of file