Moved detection of removed parameters to updateDevice function plus other organization and optimization changes
This commit is contained in:
1 parent
b41e619a89
commit
3348139795
4 files changed
+152
-143
No files matched your search
+3
-89
@@ -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
|
||||
|
||||
@@ -93,6 +93,7 @@ optimizeProjection = (obj) ->
|
||||
delete obj[b]
|
||||
keys.splice(i--, 1)
|
||||
++ i
|
||||
return
|
||||
|
||||
|
||||
exports.test = test
|
||||
|
||||
+5
-54
@@ -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) ->
|
||||
|
||||
@@ -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
|
||||
Reference in new issue
Block a user