Major overhaul to how presets are processed, needed in order to expose preset functionality through API.

This commit is contained in:
Zaid Abdulla committed 2013-10-16 19:39:20 +03:00
1 parent a22e638e9a
commit 3fbe297726
3 files changed
+299 -211

No files matched your search

+53 -25
View File
@@ -199,6 +199,58 @@ isTaskExpired = (task) ->
return false return false
assertPresets = (currentRequest) ->
db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) ->
presetsHash = results['presets_hash']
devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"]
if devicePresetsHash? and devicePresetsHash == presetsHash
# no discrepancy, return empty response
res = tr069.response(null, {})
writeResponse(currentRequest, res)
else
db.getPresets((allPresets, allObjects) ->
if not presetsHash?
presetsHash = presets.calculatePresetsHash(allPresets, allObjects)
db.memcached.set('presets_hash', presetsHash, config.PRESETS_CACHE_DURATION, (err, res) ->
)
presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, (devicePreset) ->
presets.processDevicePreset(currentRequest.deviceId, devicePreset, (taskList, addTags, deleteTags, expiry) ->
db.memcached.set("#{currentRequest.deviceId}_presets_hash", presetsHash, expiry - config.PRESETS_TIME_PADDING, (err, res) ->
)
if addTags.length + deleteTags.length + taskList.length
util.log("#{currentRequest.deviceId}: Presets discrepancy found")
if addTags.length + deleteTags.length
util.log("#{currentRequest.deviceId}: Updating tags +(#{addTags}) -(#{deleteTags})")
if deleteTags.length
db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$pull' : {'_tags' : {'$in' : deleteTags}}}, {}, (err, count) ->
throw err if err
)
if addTags.length
db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$addToSet' : {'_tags' : {'$each' : addTags}}}, {}, (err, count) ->
throw err if err
)
if taskList.length
apiFunctions.insertTasks(taskList, (err, taskList) ->
throw err if err
task = taskList[0]
util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})")
runTask(currentRequest, task, {})
)
else
res = tr069.response(null, {}, {})
writeResponse(currentRequest, res)
)
)
)
)
nextTask = (currentRequest) -> nextTask = (currentRequest) ->
cur = db.tasksCollection.find({'device' : currentRequest.deviceId}).sort(['timestamp']).limit(1) cur = db.tasksCollection.find({'device' : currentRequest.deviceId}).sort(['timestamp']).limit(1)
cur.nextObject((err, task) -> cur.nextObject((err, task) ->
@@ -207,31 +259,7 @@ nextTask = (currentRequest) ->
if not task if not task
# no more tasks, check presets discrepancy # no more tasks, check presets discrepancy
db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) -> assertPresets(currentRequest)
# ignore memcached errors
presetsHash = results['presets_hash']
devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"]
if not devicePresetsHash or presetsHash != devicePresetsHash
presets.assertPresets(currentRequest.deviceId, presetsHash, (taskList) ->
applyConfigurations(currentRequest, taskList)
)
else if not presetsHash
presets.getPresetsHash((hash) ->
if hash != devicePresetsHash
presets.assertPresets(currentRequest.deviceId, presetsHash, (taskList) ->
applyConfigurations(currentRequest, taskList)
)
else
# no discrepancy, return empty response
res = tr069.response(null, cwmpResponse)
writeResponse(currentRequest, res)
)
else
# no discrepancy, return empty response
res = tr069.response(null, cwmpResponse)
writeResponse(currentRequest, res)
)
else if task.fault? else if task.fault?
# last task was faulty. Do nothing until until task is deleted # last task was faulty. Do nothing until until task is deleted
res = tr069.response(null, cwmpResponse) res = tr069.response(null, cwmpResponse)
+29
View File
@@ -82,7 +82,36 @@ saveTask = (task, callback) ->
) )
getPresets = (callback) ->
memcached.get(['presets', 'objects'], (err, res) ->
presets = res.presets
objects = res.objects
if presets and objects
callback(presets, objects)
return
presetsCollection.find().toArray((err, p) ->
throw err if err
presets = p
memcached.set('presets', presets, config.PRESETS_CACHE_DURATION)
callback(presets, objects) if objects
)
objectsCollection.find().toArray((err, o) ->
throw err if err
objects = {}
for i in o
objects[i._id] = i
memcached.set('objects', objects, config.PRESETS_CACHE_DURATION)
callback(presets, objects) if presets
)
)
exports.memcached = memcached exports.memcached = memcached
exports.getTask = getTask exports.getTask = getTask
exports.updateTask = updateTask exports.updateTask = updateTask
exports.saveTask = saveTask exports.saveTask = saveTask
exports.getPresets = getPresets
+217 -186
View File
@@ -7,49 +7,7 @@ log = require('util').log
customCommands = require './custom-commands' customCommands = require './custom-commands'
getPresets = (callback) -> exports.calculatePresetsHash = (presets, objects) ->
db.memcached.get(['presets', 'objects'], (err, res) ->
presets = res.presets
objects = res.objects
if presets and objects
callback(presets, objects)
return
db.presetsCollection.find().toArray((err, p) ->
throw new Error(err) if err
presets = p
db.memcached.set('presets', presets, config.PRESETS_CACHE_DURATION)
callback(presets, objects) if objects
)
db.objectsCollection.find().toArray((err, o) ->
throw new Error(err) if err
objects = {}
for i in o
objects[i._id] = i
db.memcached.set('objects', objects, config.PRESETS_CACHE_DURATION)
callback(presets, objects) if presets
)
)
getPresetsHash = (callback) ->
db.memcached('presets_hash', (err, res) ->
if res
callback(res)
else
getPresets((presets, objects) ->
hash = calculatePresetsHash(presets, objects)
db.memcached.set('presets_hash', hash, config.PRESETS_CACHE_DURATION, (err, res) ->
callback(hash)
)
)
)
calculatePresetsHash = (presets, objects) ->
crypto = require('crypto') crypto = require('crypto')
hash = crypto.createHash('md5').update(JSON.stringify(presets) + JSON.stringify(objects)).digest('hex') hash = crypto.createHash('md5').update(JSON.stringify(presets) + JSON.stringify(objects)).digest('hex')
return hash return hash
@@ -64,158 +22,231 @@ matchObject = (object, param) ->
return true return true
exports.assertPresets = (deviceId, presetsHash, callback) -> exports.getDevicePreset = (deviceId, presets, objects, callback) ->
getPresets((presets, objects) -> deviceCustomCommands = customCommands.getDeviceCustomCommands(deviceId)
deviceCustomCommands = customCommands.getDeviceCustomCommands(deviceId) # only fetch relevant params
# only fetch relevant params projection = {_id : 1}
projection = {_id : 1} for p in presets
p.precondition = query.expand(p.precondition)
mongoQuery.projection(p.precondition, projection)
for c in p.configurations
switch c.type
when 'value', 'age'
projection[c.name] = 1
when 'custom_command_value', 'custom_command_age'
commandfile = c.command.split(' ', 1)
if commandfile of deviceCustomCommands
projection["_customCommands.#{c.command.split(' ', 1)}"] = 1
when 'software_version'
projection['InternetGatewayDevice.DeviceInfo.SoftwareVersion'] = 1
when 'add_tag', 'delete_tag'
projection['_tags'] = 1
when 'add_object', 'delete_object'
projection[c.name] = 1
else
throw new Error('Unknown configuration type')
mongoQuery.optimizeProjection(projection)
db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) ->
throw err if err
devicePresets = []
for p in presets for p in presets
p.precondition = query.expand(p.precondition) if mongoQuery.test(device, p.precondition)
mongoQuery.projection(p.precondition, projection) devicePresets.push(p)
for c in p.configurations configurations = accumulateConfigurations(devicePresets, objects)
switch c.type
when 'value', 'age' getParamDetails = (device, path) ->
projection[c.name] = 1 p = common.getParamValueFromPath(device, path)
when 'custom_command_value', 'custom_command_age' return undefined if not p?
commandfile = c.command.split(' ', 1) details = {}
if commandfile of deviceCustomCommands for k,v of p
projection["_customCommands.#{c.command.split(' ', 1)}"] = 1 details[k] = v if k[0] == '_'
when 'firmware' return details
projection['InternetGatewayDevice.DeviceInfo.SoftwareVersion'] = 1
when 'add_tag', 'delete_tag' devicePreset = {}
projection['_tags'] = 1 for c in configurations
when 'add_object', 'delete_object' switch c.type
projection[c.name] = 1 when 'value'
devicePreset.parameters ?= {}
devicePreset.parameters[c.name] ?= {}
devicePreset.parameters[c.name].preset ?= {}
devicePreset.parameters[c.name].preset.value = c.value
devicePreset.parameters[c.name].current = getParamDetails(device, c.name)
when 'age'
devicePreset.parameters ?= {}
devicePreset.parameters[c.name] ?= {}
devicePreset.parameters[c.name].preset ?= {}
devicePreset.parameters[c.name].preset.expiry = parseInt(c.age)
devicePreset.parameters[c.name].current = getParamDetails(device, c.name)
when 'add_tag'
devicePreset.tags ?= {}
devicePreset.tags[c.tag] = 2 | c.tag in (device._tags ? [])
when 'delete_tag'
devicePreset.tags ?= {}
devicePreset.tags[c.tag] = 0 | c.tag in (device._tags ? [])
when 'add_object'
devicePreset.objects ?= {}
devicePreset.objects[c.name] ?= {}
devicePreset.objects[c.name][c.object] ?= {}
devicePreset.objects[c.name][c.object].preset = objects[c.object]
continue if not param = common.getParamValueFromPath(device, c.name)
devicePreset.objects[c.name][c.object].current ?= {}
for k,p of param
continue if k[0] == '_'
if p._name?
if p._name == c.object
devicePreset.objects[c.name][c.object].current[k] = p
else if matchObject(objects[c.object], p)
u = {}
u["#{c.name}.#{k}._name"] = c.object
db.devicesCollection.update({'_id' : deviceId}, {'$set' : u}, {}, (err, count) ->
throw err if err
)
devicePreset.objects[c.name][c.object].current[k] = p
when 'delete_object'
devicePreset.objects ?= {}
devicePreset.objects[c.name] ?= {}
devicePreset.objects[c.name][c.object] ?= {}
continue if not param = common.getParamValueFromPath(device, c.name)
devicePreset.objects[c.name][c.object].current ?= {}
for k,p of param
continue if k[0] == '_'
if p._name?
if p._name == c.object
devicePreset.objects[c.name][c.object].current[k] = p
else if matchObject(objects[c.object], p)
u = {}
u["#{c.name}.#{k}._name"] = c.object
db.devicesCollection.update({'_id' : deviceId}, {'$set' : u}, {}, (err, count) ->
throw err if err
)
devicePreset.objects[c.name][c.object].current[k] = p
when 'custom_command_value'
[filename, commandName] = c.command.split(' ', 2)
devicePreset.customCommands ?= {}
devicePreset.customCommands[filename] ?= {}
devicePreset.customCommands[filename].preset ?= {}
devicePreset.customCommands[filename].preset.value = c.value
devicePreset.customCommands[filename].preset.valueCommand = c.valueCommand
continue if not deviceCustomCommands[filename]? or commandName not in deviceCustomCommands[filename]
if cmd = common.getParamValueFromPath(device, "_customCommands.#{filename}")
devicePreset.customCommands[filename].current = cmd
else else
throw new Error('Unknown configuration type') devicePreset.customCommands[filename].current ?= {}
mongoQuery.optimizeProjection(projection) when 'custom_command_age'
db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> [filename, commandName] = c.command.split(' ', 2)
throw new Error(err) if err? devicePreset.customCommands ?= {}
devicePresets = [] devicePreset.customCommands[filename] ?= {}
for p in presets devicePreset.customCommands[filename].preset ?= {}
if mongoQuery.test(device, p.precondition) devicePreset.customCommands[filename].preset.expiry = parseInt(c.expiry)
devicePresets.push(p) devicePreset.customCommands[filename].preset.expiryCommand = c.expiryCommand
continue if not deviceCustomCommands[filename]? or commandName not in deviceCustomCommands[filename]
configurations = accumulateConfigurations(devicePresets, objects) if cmd = common.getParamValueFromPath(device, "_customCommands.#{filename}")
now = Date.now() devicePreset.customCommands[filename].current = cmd
taskList = []
expiry = config.PRESETS_CACHE_DURATION
getParameterValues = []
setParameterValues = []
add_tags = []
delete_tags = []
for c in configurations
param = if c.name? then common.getParamValueFromPath(device, c.name) else undefined
switch c.type
when 'value'
continue if not param? # ignore parameters that don't exist
dst = common.matchType(param._value, c.value)
if param._value != dst
setParameterValues.push([c.name, dst, param._type])
when 'age'
continue if not param? # ignore parameters that don't exist
timeDiff = (now - param._timestamp) / 1000
if (c.age - timeDiff < config.PRESETS_TIME_PADDING)
expiry = Math.min(expiry, c.age)
if param._object or param._instance
taskList.push({device : deviceId, name : 'refreshObject', objectName : c.name})
else
getParameterValues.push(c.name)
else
expiry = Math.min(expiry, c.age - timeDiff)
when 'custom_command_value'
[filename, commandName] = c.command.split(' ', 2)
continue if not deviceCustomCommands[filename]? or commandName not in deviceCustomCommands[filename]
currentValue = try common.getParamValueFromPath(device, "_customCommands.#{filename}")._value catch then undefined
dst = common.matchType(currentValue, c.value)
if currentValue != dst
taskList.push({device : deviceId, name : 'customCommand', command : c.command})
when 'custom_command_age'
[filename, commandName] = c.command.split(' ', 2)
continue if not deviceCustomCommands[filename]? or commandName not in deviceCustomCommands[filename]
command = common.getParamValueFromPath(device, "_customCommands.#{filename}")
timeDiff = try (now - command._timestamp) / 1000 catch then 9999999
if (c.age - timeDiff < config.PRESETS_TIME_PADDING)
expiry = Math.min(expiry, c.age)
taskList.push({device : deviceId, name : 'customCommand', command : c.command})
else
expiry = Math.min(expiry, c.age - timeDiff)
when 'add_tag'
add_tags.push(c.tag) if not device['_tags']? or c.tag not in device['_tags']
when 'delete_tag'
delete_tags.push(c.tag) if device['_tags']? and c.tag in device['_tags']
when 'add_object'
instances = {}
for k,p of param
continue if k[0] == '_'
if p._name?
if p._name == c.object
instances[k] = p
else if matchObject(objects[c.object], p)
u = {}
u["#{c.name}.#{k}._name"] = c.object
db.devicesCollection.update({'_id' : deviceId}, {'$set' : u}, {}, (err, count) ->
throw new Error(err) if err?
)
instances[k] = p
if Object.keys(instances).length > 0
for k,i of instances
for k2,j of objects[c.object]
continue if k2[0] == '_'
dst = common.matchType(param[k][k2]._value, j)
if param[k][k2]._value != dst
setParameterValues.push(["#{c.name}.#{k}.#{k2}", dst, param[k][k2]._value])
else
vals = []
for k2,j of objects[c.object]
vals.push([k2, j]) if k2[0] != '_'
taskList.push({device : deviceId, name : 'addObject', objectName : c.name, parameterValues : vals, instanceName : c.object})
when 'delete_object'
for k,p of param
continue if k[0] == '_'
if p._name?
if p._name == c.object
taskList.push({device : deviceId, name : 'deleteObject', objectName : "#{c.name}.#{k}"})
else if matchObject(objects[c.object], p)
taskList.push({device : deviceId, name : 'deleteObject', objectName : "#{c.name}.#{k}"})
else else
throw new Error('Unknown configuration type') devicePreset.customCommands[filename].current ?= {}
if add_tags.length + delete_tags.length > 0 when 'software_version'
log("#{deviceId}: Updating tags") devicePreset.softwareVersion ?= {}
devicePreset.softwareVersion.preset = c.softwareVersion
devicePreset.softwareVersion.current = common.getParamValueFromPath(device, 'InternetGatewayDevice.DeviceInfo.SoftwareVersion')
if delete_tags.length > 0 else
db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : {'$in' : delete_tags}}}, {}, (err, count) -> throw new Error('Unknown configuration type')
throw new Error(err) if err?
)
if add_tags.length > 0 callback(devicePreset)
db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : {'$each' : add_tags}}}, {}, (err, count) ->
throw new Error(err) if err?
)
if getParameterValues.length
taskList.push {device : deviceId, name : 'getParameterValues', parameterNames: getParameterValues, timestamp : new Date()}
if setParameterValues.length
taskList.push {device : deviceId, name : 'setParameterValues', parameterValues: setParameterValues, timestamp : new Date()}
if not presetsHash
presetsHash = calculatePresetsHash(presets, objects)
db.memcached.set('presets_hash', presetsHash, config.PRESETS_CACHE_DURATION, (err, res) ->
)
db.memcached.set("#{deviceId}_presets_hash", presetsHash, expiry - config.PRESETS_TIME_PADDING, (err, res) ->
callback(taskList)
)
)
) )
exports.processDevicePreset = (deviceId, devicePreset, callback) ->
now = new Date()
expiry = config.PRESETS_CACHE_DURATION
getParameterValues = []
setParameterValues = []
addTags = []
deleteTags = []
taskList = []
# Parameters
for parameterPath, parameterDetails of devicePreset.parameters ? {}
continue if not parameterDetails.current?
expiry = Math.min(expiry, parameterDetails.preset.expiry) if parameterDetails.preset.expiry?
presetValue = parameterDetails.preset.value
currentValue = parameterDetails.current._value
if presetValue? and currentValue != (presetValue = common.matchType(currentValue, presetValue))
setParameterValues.push([parameterPath, presetValue, parameterDetails.current._type])
else if parameterDetails.preset.expiry?
diff = (now - (parameterDetails.current._timestamp ? 0)) / 1000
if diff >= expiry
if parameterDetails.current._value?
getParameterValues.push(parameterPath)
else
taskList.push({device : deviceId, name : 'refreshObject', objectName : parameterPath})
else
expiry = Math.min(expiry, diff)
# Objects
for parameterPath, object of devicePreset.objects ? {}
for objectName, objectDetails of object
continue if not objectDetails.current?
if objectDetails.preset?
if Object.keys(objectDetails.current) > 0
for i, obj of objectDetails.current
for paramName, paramDetails of obj
currentValue = paramDetails.value
presetValue = common.matchType(currentValue, objectDetails.preset[paramName]?.value)
if presetValue? and currentValue != presetValue
setParameterValues.push(["#{parameterPath}.#{i}.#{paramName}", presetValue, paramDetails.type])
else
vals = []
for k,v of allObjects[objectName]
vals.push([k, v]) if k[0] != '_'
taskList.push({device : deviceId, name : 'addObject', objectName : parameterPath, parameterValues : vals, instanceName : objectName})
else if Object.keys(objectDetails.current) > 0
for i, obj of objectDetails.current
taskList.push({device : deviceId, name : 'deleteObject', objectName : "#{parameterPath}.#{i}"})
# Tags
for tag, v of devicePreset.tags ? {}
if v == 1
deleteTags.push(tag)
else if v == 2
addTags.push(tag)
# Custom commands
for commandName, commandDetails of devicePreset.customCommands ? {}
continue if not commandDetails.current?
expiry = Math.min(expiry, commandDetails.preset.expiry) if commandDetails.preset.expiry?
currentValue = commandDetails.current._value
presetValue = common.matchType(currentValue, commandDetails.preset.value)
if commandDetails.preset.value? and currentValue != presetValue
taskList.push({device : deviceId, name : 'customCommand', command : commandDetails.preset.valueCommand})
else if commandDetails.preset.expiry?
diff = (now - (commandDetails.current._timestamp ? 0)) / 1000
if diff <= 0
taskList.push({device : deviceId, name : 'customCommand', command : commandDetails.preset.expiryCommand})
else
expiry = Math.min(expiry, diff)
if setParameterValues.length
taskList.push {device : deviceId, name : 'setParameterValues', parameterValues: setParameterValues, timestamp : now}
if getParameterValues.length
taskList.push {device : deviceId, name : 'getParameterValues', parameterNames: getParameterValues, timestamp : now}
callback(taskList, addTags, deleteTags, expiry)
getObjectHash = (object) -> getObjectHash = (object) ->
return object._id if not object._keys? or object._keys.length == 0 return object._id if not object._keys? or object._keys.length == 0
hash = '' hash = ''
@@ -235,8 +266,8 @@ accumulateConfigurations = (presets, objects) ->
when 'add_object', 'delete_object' when 'add_object', 'delete_object'
objectHash = getObjectHash(objects[c.object]) objectHash = getObjectHash(objects[c.object])
"object_#{c.name}_#{objectHash}" "object_#{c.name}_#{objectHash}"
when 'firmware' when 'software_version'
'firmware' 'software_version'
else else
"#{c.type}_#{c.name}" "#{c.type}_#{c.name}"