Renamed profiles to presets. Added preconditition.
This commit is contained in:
1 parent
5d95f6039b
commit
04b92945fe
7 files changed
+230
-143
No files matched your search
+12
-12
@@ -6,7 +6,7 @@ tr069 = require './tr-069'
|
|||||||
tasks = require './tasks'
|
tasks = require './tasks'
|
||||||
sanitize = require('./sanitize').sanitize
|
sanitize = require('./sanitize').sanitize
|
||||||
db = require './db'
|
db = require './db'
|
||||||
profiles = require './profiles'
|
presets = require './presets'
|
||||||
|
|
||||||
|
|
||||||
currentRequest = null
|
currentRequest = null
|
||||||
@@ -14,7 +14,7 @@ currentRequest = null
|
|||||||
|
|
||||||
applyConfigurations = (taskList) ->
|
applyConfigurations = (taskList) ->
|
||||||
if taskList.length
|
if taskList.length
|
||||||
util.log("#{currentRequest.deviceId}: Profiles discrepancy found")
|
util.log("#{currentRequest.deviceId}: Presets discrepancy found")
|
||||||
db.tasksCollection.save(taskList, (err) ->
|
db.tasksCollection.save(taskList, (err) ->
|
||||||
for task in taskList
|
for task in taskList
|
||||||
util.log("#{currentRequest.deviceId}: Added #{task.name} task #{task._id}")
|
util.log("#{currentRequest.deviceId}: Added #{task.name} task #{task._id}")
|
||||||
@@ -120,19 +120,19 @@ nextTask = (cookies) ->
|
|||||||
resParams = {}
|
resParams = {}
|
||||||
|
|
||||||
if not task
|
if not task
|
||||||
# no more taks, check profile discrepancy
|
# no more taks, check presets discrepancy
|
||||||
db.memcached.get(["#{currentRequest.deviceId}_profiles_hash", 'profiles_hash'], (err, results) ->
|
db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) ->
|
||||||
profilesHash = results['profiles_hash']
|
presetsHash = results['presets_hash']
|
||||||
deviceProfilesHash = results["#{currentRequest.deviceId}_profiles_hash"]
|
devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"]
|
||||||
|
|
||||||
if not deviceProfilesHash or profilesHash != deviceProfilesHash
|
if not devicePresetsHash or presetsHash != devicePresetsHash
|
||||||
profiles.findDeviceConfigurationDiscrepancy(currentRequest.deviceId, profilesHash, (taskList) ->
|
presets.assertPresets(currentRequest.deviceId, presetsHash, (taskList) ->
|
||||||
applyConfigurations(taskList)
|
applyConfigurations(taskList)
|
||||||
)
|
)
|
||||||
else if not profilesHash
|
else if not presetsHash
|
||||||
profiles.getProfilesHash((hash) ->
|
presets.getPresetsHash((hash) ->
|
||||||
if hash != deviceProfilesHash
|
if hash != devicePresetsHash
|
||||||
profiles.findDeviceConfigurationDiscrepancy(currentRequest.deviceId, profilesHash, (taskList) ->
|
presets.assertPresets(currentRequest.deviceId, presetsHash, (taskList) ->
|
||||||
applyConfigurations(taskList)
|
applyConfigurations(taskList)
|
||||||
)
|
)
|
||||||
else
|
else
|
||||||
|
|||||||
@@ -1,3 +1,14 @@
|
|||||||
|
exports.UNDEFINED_TYPE = '[object Undefined]'
|
||||||
|
exports.NULL_TYPE = '[object Null]'
|
||||||
|
exports.NUMBER_TYPE = '[object Number]'
|
||||||
|
exports.STRING_TYPE = '[object String]'
|
||||||
|
exports.ARRAY_TYPE = '[object Array]'
|
||||||
|
exports.OBJECT_TYPE = '[object Object]'
|
||||||
|
|
||||||
|
exports.typeOf = (obj) ->
|
||||||
|
Object.prototype.toString.call(obj)
|
||||||
|
|
||||||
|
|
||||||
exports.endsWith = (str, suffix) ->
|
exports.endsWith = (str, suffix) ->
|
||||||
str.indexOf(suffix, str.length - suffix.length) isnt -1
|
str.indexOf(suffix, str.length - suffix.length) isnt -1
|
||||||
|
|
||||||
@@ -16,7 +27,15 @@ exports.getDeviceId = (deviceIdStruct) ->
|
|||||||
|
|
||||||
return "#{escape(deviceIdStruct['OUI'])}-#{escape(deviceIdStruct['SerialNumber'])}"
|
return "#{escape(deviceIdStruct['OUI'])}-#{escape(deviceIdStruct['SerialNumber'])}"
|
||||||
|
|
||||||
|
|
||||||
exports.extend = (obj, mixin) ->
|
exports.extend = (obj, mixin) ->
|
||||||
obj[name] = method for name, method of mixin
|
obj[name] = method for name, method of mixin
|
||||||
obj
|
obj
|
||||||
|
|
||||||
|
|
||||||
|
exports.getParamValueFromPath = (obj, path) ->
|
||||||
|
pp = path.split('.')
|
||||||
|
ref = obj
|
||||||
|
for p in pp
|
||||||
|
ref = ref[p]
|
||||||
|
return ref
|
||||||
+2
-2
@@ -4,8 +4,8 @@ exports.ACS_INTERFACE = '0.0.0.0'
|
|||||||
exports.API_PORT = 7557
|
exports.API_PORT = 7557
|
||||||
exports.API_INTERFACE = '0.0.0.0'
|
exports.API_INTERFACE = '0.0.0.0'
|
||||||
exports.CACHE_DURATION = 60 # in seconds
|
exports.CACHE_DURATION = 60 # in seconds
|
||||||
exports.PROFILES_CACHE_DURATION = 86400
|
exports.PRESETS_CACHE_DURATION = 86400
|
||||||
exports.PROFILES_TIME_PADDING = 1
|
exports.PRESETS_TIME_PADDING = 1
|
||||||
exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock'
|
exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock'
|
||||||
exports.MEMCACHED_SOCKET = '/tmp/memcached.sock'
|
exports.MEMCACHED_SOCKET = '/tmp/memcached.sock'
|
||||||
exports.WORKER_RESPAWN_TIME = 3000
|
exports.WORKER_RESPAWN_TIME = 3000
|
||||||
|
|||||||
@@ -4,10 +4,6 @@ Memcached = require 'memcached'
|
|||||||
Memcached.config.maxExpiration = 86400
|
Memcached.config.maxExpiration = 86400
|
||||||
memcached = new Memcached(config.MEMCACHED_SOCKET)
|
memcached = new Memcached(config.MEMCACHED_SOCKET)
|
||||||
|
|
||||||
# Create MongoDB connections
|
|
||||||
tasksCollection = null
|
|
||||||
devicesCollection = null
|
|
||||||
profilesCollection = null
|
|
||||||
|
|
||||||
dbserver = new mongo.Server(config.MONGODB_SOCKET, 0, {auto_reconnect: true})
|
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 = new mongo.Db(config.DATABASE_NAME, dbserver, {native_parser:true, safe:true})
|
||||||
@@ -23,8 +19,8 @@ db.open( (err, db) ->
|
|||||||
exports.devicesCollection = collection
|
exports.devicesCollection = collection
|
||||||
)
|
)
|
||||||
|
|
||||||
db.collection('profiles', (err, collection) ->
|
db.collection('presets', (err, collection) ->
|
||||||
exports.profilesCollection = collection
|
exports.presetsCollection = collection
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,80 @@
|
|||||||
|
common = require './common'
|
||||||
|
|
||||||
|
|
||||||
|
test = (obj, query) ->
|
||||||
|
for k,v of query
|
||||||
|
if k.charAt(0) == '$' # this is a logical operator
|
||||||
|
res = testExpressions(obj, v, k)
|
||||||
|
else
|
||||||
|
value = common.getParamValueFromPath(obj, k)
|
||||||
|
|
||||||
|
if common.typeOf(v) isnt common.OBJECT_TYPE
|
||||||
|
if common.typeOf(value) is common.ARRAY_TYPE
|
||||||
|
res = value.indexOf(v) != -1
|
||||||
|
else
|
||||||
|
res = v == value
|
||||||
|
else
|
||||||
|
for k2,v2 of v
|
||||||
|
switch k2
|
||||||
|
when '$ne'
|
||||||
|
if common.typeOf(value) is common.ARRAY_TYPE
|
||||||
|
res = value.indexOf(v2) == -1
|
||||||
|
else
|
||||||
|
res = value != v2
|
||||||
|
when '$lt'
|
||||||
|
res = value < v2
|
||||||
|
when '$lte'
|
||||||
|
res = value <= v2
|
||||||
|
when '$gt'
|
||||||
|
res = value > v2
|
||||||
|
when '$gte'
|
||||||
|
res = value >= v2
|
||||||
|
when '$regex'
|
||||||
|
res = v2.test(value)
|
||||||
|
when '$in'
|
||||||
|
throw new Error('Operator not supported')
|
||||||
|
when '$nin'
|
||||||
|
throw new Error('Operator not supported')
|
||||||
|
when '$all'
|
||||||
|
throw new Error('Operator not supported')
|
||||||
|
when '$exists'
|
||||||
|
throw new Error('Operator not supported')
|
||||||
|
else
|
||||||
|
throw new Error('Operator not supported')
|
||||||
|
|
||||||
|
if not res
|
||||||
|
return false
|
||||||
|
|
||||||
|
return true
|
||||||
|
|
||||||
|
|
||||||
|
testExpressions = (obj, expressions, lop) ->
|
||||||
|
for f in expressions
|
||||||
|
res = test(obj, f)
|
||||||
|
|
||||||
|
switch lop
|
||||||
|
when '$and'
|
||||||
|
return false if not res
|
||||||
|
when '$or'
|
||||||
|
return true if res
|
||||||
|
when '$nor'
|
||||||
|
return false if res
|
||||||
|
|
||||||
|
return switch lop
|
||||||
|
when '$and' then true
|
||||||
|
when '$or' then false
|
||||||
|
when '$nor' then true
|
||||||
|
else throw new Error('Unknown logical operator')
|
||||||
|
|
||||||
|
|
||||||
|
projection = (query, proj) ->
|
||||||
|
for k,v of query
|
||||||
|
if k.charAt(0) == '$' # this is a logical operator
|
||||||
|
for q in v
|
||||||
|
projection(q, proj)
|
||||||
|
else
|
||||||
|
proj[k] = 1
|
||||||
|
|
||||||
|
|
||||||
|
exports.test = test
|
||||||
|
exports.projection = projection
|
||||||
+115
@@ -0,0 +1,115 @@
|
|||||||
|
config = require './config'
|
||||||
|
common = require './common'
|
||||||
|
db = require './db'
|
||||||
|
mongoQuery = require './mongo-query'
|
||||||
|
|
||||||
|
|
||||||
|
getPresets = (callback) ->
|
||||||
|
db.memcached.get('presets', (err, res) ->
|
||||||
|
throw new Error(err) if err
|
||||||
|
|
||||||
|
if res
|
||||||
|
callback(res)
|
||||||
|
return
|
||||||
|
|
||||||
|
db.presetsCollection.find().toArray((err, presets) ->
|
||||||
|
throw new Error(err) if err
|
||||||
|
|
||||||
|
db.memcached.set('presets', presets, config.PRESETS_CACHE_DURATION)
|
||||||
|
callback(presets)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
getPresetsHash = (callback) ->
|
||||||
|
db.memcached('presets_hash', (err, res) ->
|
||||||
|
if res
|
||||||
|
callback(res)
|
||||||
|
else
|
||||||
|
getPresets((presets) ->
|
||||||
|
hash = callback(calculatePresetsHash(presets))
|
||||||
|
db.memcached.set('presets_hash', hash, config.PRESETS_CACHE_DURATION, (err, res) ->
|
||||||
|
callback(hash)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
calculatePresetsHash = (presets) ->
|
||||||
|
crypto = require('crypto')
|
||||||
|
hash = crypto.createHash('md5').update(JSON.stringify(presets)).digest('hex')
|
||||||
|
return hash
|
||||||
|
|
||||||
|
|
||||||
|
exports.assertPresets = (deviceId, presetsHash, callback) ->
|
||||||
|
getPresets((presets) ->
|
||||||
|
# only fetch relevant params
|
||||||
|
projection = {}
|
||||||
|
for p in presets
|
||||||
|
mongoQuery.projection(p.precondition, projection)
|
||||||
|
|
||||||
|
for c in p.configurations
|
||||||
|
switch c.type
|
||||||
|
when 'value', 'age'
|
||||||
|
projection[c.name] = 1
|
||||||
|
when 'firmware'
|
||||||
|
projection['InternetGatewayDevice.DeviceInfo.SoftwareVersion'] = 1
|
||||||
|
else
|
||||||
|
throw new Error('Unknown configuration type')
|
||||||
|
|
||||||
|
db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) ->
|
||||||
|
devicePresets = []
|
||||||
|
for p in presets
|
||||||
|
if mongoQuery.test(device, p.precondition)
|
||||||
|
devicePresets.push(p)
|
||||||
|
|
||||||
|
configurations = accumulateConfigurations(devicePresets)
|
||||||
|
now = Date.now()
|
||||||
|
taskList = []
|
||||||
|
expiry = config.PRESETS_CACHE_DURATION
|
||||||
|
getParameterValues = []
|
||||||
|
setParameterValues = []
|
||||||
|
for c in configurations
|
||||||
|
switch c.type
|
||||||
|
when 'value'
|
||||||
|
if c.value != common.getParamValueFromPath(device, "#{c.name}._value")
|
||||||
|
setParameterValues.push([c.name, c.value])
|
||||||
|
when 'age'
|
||||||
|
timeDiff = (now - common.getParamValueFromPath(device, "#{c.name}._timestamp")) / 1000
|
||||||
|
if (c.age - timeDiff < config.PRESETS_TIME_PADDING)
|
||||||
|
expiry = Math.min(expiry, c.age)
|
||||||
|
getParameterValues.push(c.name)
|
||||||
|
else
|
||||||
|
expiry = Math.min(expiry, c.age - timeDiff)
|
||||||
|
else
|
||||||
|
throw new Error('Unknown configuration type')
|
||||||
|
if getParameterValues.length
|
||||||
|
taskList.push {device : deviceId, name : 'getParameterValues', parameterNames: getParameterValues, timestamp : db.mongo.Timestamp()}
|
||||||
|
if setParameterValues.length
|
||||||
|
taskList.push {device : deviceId, name : 'setParameterValues', parameterValues: getParameterValues, timestamp : db.mongo.Timestamp()}
|
||||||
|
|
||||||
|
if not presetsHash
|
||||||
|
presetsHash = calculatePresetsHash(presets)
|
||||||
|
db.memcached.set('presets_hash', presetsHash, expiry - config.PRESETS_TIME_PADDING, (err, res) ->
|
||||||
|
)
|
||||||
|
|
||||||
|
db.memcached.set("#{deviceId}_presets_hash", presetsHash, expiry - config.PRESETS_TIME_PADDING, (err, res) ->
|
||||||
|
callback(taskList)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
accumulateConfigurations = (presets) ->
|
||||||
|
maxWeights = {}
|
||||||
|
configurations = {}
|
||||||
|
for p in presets
|
||||||
|
for c in p.configurations
|
||||||
|
configurationHash = if c.name? then "#{c.type}_#{c.name}" else c.type
|
||||||
|
|
||||||
|
if not maxWeights[configurationHash]? or p.weight > maxWeights[configurationHash]
|
||||||
|
configurations[configurationHash] = c
|
||||||
|
maxWeights[configurationHash] = p.weight
|
||||||
|
|
||||||
|
configurationsList = (configurations[c] for c of configurations)
|
||||||
|
return configurationsList
|
||||||
-123
@@ -1,123 +0,0 @@
|
|||||||
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
|
|
||||||
|
|
||||||
|
|
||||||
exports.findDeviceConfigurationDiscrepancy = (deviceId, profilesHash, callback) ->
|
|
||||||
getProfiles((profiles) ->
|
|
||||||
params = {'_tags' : 1}
|
|
||||||
for p of profiles
|
|
||||||
profile = profiles[p]
|
|
||||||
for c in profile.configurations
|
|
||||||
if c.type == 'get' or c.type == 'set'
|
|
||||||
params[c.name] = 1
|
|
||||||
|
|
||||||
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)
|
|
||||||
now = Date.now()
|
|
||||||
taskList = []
|
|
||||||
expiry = config.PROFILES_CACHE_DURATION
|
|
||||||
getParameterValues = []
|
|
||||||
setParameterValues = []
|
|
||||||
for c in combinedConfigurations
|
|
||||||
switch c.type
|
|
||||||
when 'set'
|
|
||||||
if c.value != getDeviceParamFromPath(device, "#{c.name}._value")
|
|
||||||
setParameterValues.push([c.name, c.value])
|
|
||||||
when 'get'
|
|
||||||
timeDiff = (now - getDeviceParamFromPath(device, "#{c.name}._timestamp")) / 1000
|
|
||||||
if (c.age - timeDiff < config.PROFILES_TIME_PADDING)
|
|
||||||
expiry = Math.min(expiry, c.age)
|
|
||||||
getParameterValues.push(c.name)
|
|
||||||
else
|
|
||||||
expiry = Math.min(expiry, c.age - timeDiff)
|
|
||||||
else
|
|
||||||
throw new Error('Unknown configuration type')
|
|
||||||
if getParameterValues.length
|
|
||||||
taskList.push {device : deviceId, name : 'getParameterValues', parameterNames: getParameterValues, timestamp : db.mongo.Timestamp()}
|
|
||||||
if setParameterValues.length
|
|
||||||
taskList.push {device : deviceId, name : 'setParameterValues', parameterValues: getParameterValues, timestamp : db.mongo.Timestamp()}
|
|
||||||
|
|
||||||
if not profilesHash
|
|
||||||
profilesHash = calculateProfilesHash(profiles)
|
|
||||||
db.memcached.set('profiles_hash', profilesHash, expiry - config.PROFILES_TIME_PADDING, (err, res) ->
|
|
||||||
)
|
|
||||||
|
|
||||||
db.memcached.set("#{deviceId}_profiles_hash", profilesHash, expiry - config.PROFILES_TIME_PADDING, (err, res) ->
|
|
||||||
callback(taskList)
|
|
||||||
)
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
combineProfileConfigurations = (profiles) ->
|
|
||||||
maxWeights = {}
|
|
||||||
configurations = {}
|
|
||||||
for p in profiles
|
|
||||||
for c in p.configurations
|
|
||||||
configurationHash = if c.name? then "#{c.type}_#{c.name}" else c.type
|
|
||||||
|
|
||||||
if not maxWeights[configurationHash]? or p.weight > maxWeights[configurationHash]
|
|
||||||
configurations[configurationHash] = c
|
|
||||||
maxWeights[configurationHash] = p.weight
|
|
||||||
|
|
||||||
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
|
|
||||||
Reference in new issue
Block a user