Virtual parameters

Implemented support for user-defined parameters driven by Javascript
code. Renamed scripts collection to provisions.
This commit is contained in:
Zaid Abdulla committed 2016-04-17 18:54:48 +03:00
1 parent 359ff44819
commit 7b18cb3bbf
6 files changed
+487 -252

No files matched your search

+68 -25
View File
@@ -25,7 +25,8 @@ REFRESH = 3000
nextRefresh = 0 nextRefresh = 0
hash = null hash = null
presets = null presets = null
scripts = null provisions = null
virtualParameters = null
UNLOCK_SCRIPT = 'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end' UNLOCK_SCRIPT = 'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end'
@@ -55,18 +56,22 @@ lock = (lockName, ttl, callback) ->
computeHash = () -> computeHash = () ->
# MD5 hash for presets and scripts for detecting changes # MD5 hash for presets, provisions, virtual parameters for detecting changes
h = crypto.createHash('md5') h = crypto.createHash('md5')
for p in presets for p in presets
h.update(JSON.stringify(p.name)) h.update(JSON.stringify(p.name))
h.update(JSON.stringify(p.precondition)) h.update(JSON.stringify(p.precondition))
h.update(JSON.stringify(p.provisions)) h.update(JSON.stringify(p.provisions))
for s in Object.keys(scripts).sort() keys = Object.keys(provisions).sort()
h.update(JSON.stringify(s)) h.update(JSON.stringify(keys))
h.update(JSON.stringify(scripts[s].source)) h.update(provisions[k].md5) for k in keys
return h.digest('hex') keys = Object.keys(virtualParameters).sort()
h.update(JSON.stringify(keys))
h.update(virtualParameters[k].md5) for k in keys
hash = h.digest('hex')
refresh = (callback) -> refresh = (callback) ->
@@ -79,7 +84,7 @@ refresh = (callback) ->
return callback() return callback()
lock('presets_hash_lock', 3000, (err, unlockOrExtend) -> lock('presets_hash_lock', 3000, (err, unlockOrExtend) ->
counter = 2 counter = 3
db.presetsCollection.find().toArray((err, res) -> db.presetsCollection.find().toArray((err, res) ->
if err if err
callback(err) if -- counter >= 0 callback(err) if -- counter >= 0
@@ -98,25 +103,25 @@ refresh = (callback) ->
for preset in res for preset in res
precondition = JSON.parse(preset.precondition) precondition = JSON.parse(preset.precondition)
provisions = preset.provisions or [] _provisions = preset.provisions or []
# Generate provisions from the old configuration format # Generate provisions from the old configuration format
for c in preset.configurations for c in preset.configurations
switch c.type switch c.type
when 'age' when 'age'
provisions.push(['refresh', c.name, c.age * -1000]) _provisions.push(['refresh', c.name, c.age * -1000])
when 'value' when 'value'
provisions.push(['value', c.name, c.value]) _provisions.push(['value', c.name, c.value])
when 'add_tag' when 'add_tag'
provisions.push(['tag', c.tag, true]) _provisions.push(['tag', c.tag, true])
when 'delete_tag' when 'delete_tag'
provisions.push(['tag', c.tag, false]) _provisions.push(['tag', c.tag, false])
when 'script' when 'provision'
provisions.push([c.name].concat(c.args or [])) _provisions.push([c.name].concat(c.args or []))
else else
throw new Error("Unknown configuration type #{c.type}") throw new Error("Unknown configuration type #{c.type}")
presets.push({name: preset._id, precondition: precondition, provisions: provisions}) presets.push({name: preset._id, precondition: precondition, provisions: _provisions})
if -- counter == 0 if -- counter == 0
computeHash() computeHash()
@@ -127,18 +132,38 @@ refresh = (callback) ->
) )
) )
db.scriptsCollection.find().toArray((err, res) -> db.provisionsCollection.find().toArray((err, res) ->
if err if err
callback(err) if -- counter >= 0 callback(err) if -- counter >= 0
counter = 0 counter = 0
return return
scripts = {} provisions = {}
for r in res for r in res
scripts[r._id] = {} provisions[r._id] = {}
scripts[r._id].source = r.source provisions[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex')
scripts[r._id].md5 = r.md5 provisions[r._id].script = vm.Script(r.script, {filename: r._id, timeout: 50})
scripts[r._id].compiled = vm.Script(r.source, {filename: r._id, timeout: 50})
if -- counter == 0
computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0)
nextRefresh = now + REFRESH
return callback()
)
)
db.virtualParametersCollection.find().toArray((err, res) ->
if err
callback(err) if -- counter >= 0
counter = 0
return
virtualParameters = {}
for r in res
virtualParameters[r._id] = {}
virtualParameters[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex')
virtualParameters[r._id].script = vm.Script(r.script, {filename: r._id, timeout: 50})
if -- counter == 0 if -- counter == 0
computeHash() computeHash()
@@ -162,14 +187,32 @@ getPresets = (callback) ->
) )
getScripts = (callback) -> getProvisions = (callback) ->
if Date.now() < nextRefresh if Date.now() < nextRefresh
return callback(null, hash, scripts) return callback(null, hash, provisions)
refresh((err) -> refresh((err) ->
return callback(err, hash, scripts) return callback(err, hash, provisions)
) )
getVirtualParameters = (callback) ->
if Date.now() < nextRefresh
return callback(null, hash, virtualParameters)
refresh((err) ->
return callback(err, hash, virtualParameters)
)
getProvisionsAndVirtualParameters = (callback) ->
if Date.now() < nextRefresh
return callback(null, hash, provisions, virtualParameters)
refresh((err) ->
return callback(err, hash, provisions, virtualParameters)
)
exports.getPresets = getPresets exports.getPresets = getPresets
exports.getScripts = getScripts exports.getProvisions = getProvisions
exports.getProvisionsAndVirtualParameters = getProvisionsAndVirtualParameters
+12 -4
View File
@@ -26,11 +26,12 @@ tasksCollection = null
devicesCollection = null devicesCollection = null
presetsCollection = null presetsCollection = null
objectsCollection = null objectsCollection = null
scriptsCollection = null provisionsCollection = null
virtualParametersCollection = null
connect = (callback) -> connect = (callback) ->
callbackCounter = 7 callbackCounter = 8
mongodb.MongoClient.connect(config.get('MONGODB_CONNECTION_URL'), {db:{w:1},server:{autoReconnect:true}}, (err, db) -> mongodb.MongoClient.connect(config.get('MONGODB_CONNECTION_URL'), {db:{w:1},server:{autoReconnect:true}}, (err, db) ->
return callback(err) if err return callback(err) if err
exports.mongoDb = db exports.mongoDb = db
@@ -72,8 +73,15 @@ connect = (callback) ->
return callback(err) return callback(err)
) )
db.collection('scripts', (err, collection) -> db.collection('provisions', (err, collection) ->
exports.scriptsCollection = scriptsCollection = collection exports.provisionsCollection = provisionsCollection = collection
if --callbackCounter == 0 or err
callbackCounter = 0
return callback(err)
)
db.collection('virtualParameters', (err, collection) ->
exports.virtualParametersCollection = virtualParametersCollection = collection
if --callbackCounter == 0 or err if --callbackCounter == 0 or err
callbackCounter = 0 callbackCounter = 0
return callback(err) return callback(err)
+44 -2
View File
@@ -98,7 +98,7 @@ init = () ->
declare = (deviceData, path, declaration, now) -> declare = (deviceData, path, declaration, now) ->
wildcard = false wildcard = false
for i in [0...path.length] when not path[i]? for i in [0...path.length] by 1 when not path[i]?
wildcard = true wildcard = true
break break
@@ -137,7 +137,7 @@ declare = (deviceData, path, declaration, now) ->
set = (deviceData, path, revision, properties) -> set = (deviceData, path, revision, properties) ->
++ revision # revisions start at layer 1 ++ revision # revisions start at layer 1
wildcard = false wildcard = false
for i in [0...path.length] when not path[i]? for i in [0...path.length] by 1 when not path[i]?
wildcard = true wildcard = true
break break
@@ -343,6 +343,45 @@ traverse = (deviceData, pattern, revision, callback) ->
return recursive(deviceData, [], [], 0, 1) return recursive(deviceData, [], [], 0, 1)
collapse = (deviceData, revision, wildcard) ->
ref = deviceData
if ref[0]?
for depth, i in ref[0] when depth?.length > revision + 2
if i == 0 and not wildcard
depth[revision + 1] ?= []
for j in [revision + 2...depth.length] by 1 when depth[j]?
for k in [0...depth[j].length] by 2 when depth[j][k]?
depth[revision + 1][k] = depth[j][k]
depth[revision + 1][k + 1] = depth[j][k + 1]
else
for j in [revision + 2...depth.length] by 1 when depth[j]?
depth[revision + 1] = depth[j]
depth.length = revision + 2
for i in [1...ref.length] by 1 when ref[i]?
for k, v of ref[i]
collapse(v, revision, wildcard or i != 1)
clearDeclarations = (deviceData, prefix) ->
ref = deviceData
if prefix?
for p in prefix
ref = ref[1]?[p]
return if not ref?
if ref[0]?
for depth in ref[0] when depth?
depth[0] = null
for i in [1...ref.length] by 1 when ref[i]?
for k, v of ref[i]
clearDeclarations(v)
getPrerequisiteDeclarations = (declarations) -> getPrerequisiteDeclarations = (declarations) ->
dec = [declarations[0], 1] dec = [declarations[0], 1]
for i in [1...declarations.length] by 2 for i in [1...declarations.length] by 2
@@ -368,3 +407,6 @@ exports.getAll = getAll
exports.declare = declare exports.declare = declare
exports.set = set exports.set = set
exports.traverse = traverse exports.traverse = traverse
exports.collapse = collapse
exports.clearDeclarations = clearDeclarations
exports.sanitizeParameterValue = sanitizeParameterValue
+42 -9
View File
@@ -41,7 +41,6 @@
url = require 'url' url = require 'url'
mongodb = require 'mongodb' mongodb = require 'mongodb'
querystring = require 'querystring' querystring = require 'querystring'
crypto = require 'crypto'
config = require './config' config = require './config'
common = require './common' common = require './common'
@@ -59,7 +58,8 @@ FILES_REGEX = /^\/files\/([a-zA-Z0-9\%\!\*\'\(\)\;\:\@\&\=\+\$\,\?\#\[\]\-\_\.\~
PING_REGEX = /^\/ping\/([a-zA-Z0-9\-\_\.]+)\/?$/ PING_REGEX = /^\/ping\/([a-zA-Z0-9\-\_\.]+)\/?$/
QUERY_REGEX = /^\/([a-zA-Z0-9_]+s)\/?$/ QUERY_REGEX = /^\/([a-zA-Z0-9_]+s)\/?$/
DELETE_DEVICE_REGEX = /^\/devices\/([a-zA-Z0-9\-\_\%]+)\/?$/ DELETE_DEVICE_REGEX = /^\/devices\/([a-zA-Z0-9\-\_\%]+)\/?$/
SCRIPTS_REGEX = /^\/scripts\/([a-zA-Z0-9\-\_\%]+)\/?$/ PROVISIONS_REGEX = /^\/provisions\/([a-zA-Z0-9\-\_\%]+)\/?$/
VIRTUAL_PARAMETERS_REGEX = /^\/virtual_parameters\/([a-zA-Z0-9\-\_\%]+)\/?$/
errorToString = (err) -> errorToString = (err) ->
@@ -151,16 +151,15 @@ listener = (request, response) ->
else else
response.writeHead 405, {'Allow': 'PUT, DELETE'} response.writeHead 405, {'Allow': 'PUT, DELETE'}
response.end('405 Method Not Allowed') response.end('405 Method Not Allowed')
else if SCRIPTS_REGEX.test(urlParts.pathname) else if PROVISIONS_REGEX.test(urlParts.pathname)
scriptName = querystring.unescape(SCRIPTS_REGEX.exec(urlParts.pathname)[1]) provisionName = querystring.unescape(PROVISIONS_REGEX.exec(urlParts.pathname)[1])
if request.method == 'PUT' if request.method == 'PUT'
object = { object = {
_id: scriptName _id: provisionName
source: body.toString() script: body.toString()
md5: crypto.createHash('md5').update(body).digest('hex')
} }
db.scriptsCollection.save(object, (err) -> db.provisionsCollection.save(object, (err) ->
db.redisClient.del('presets_hash', (err) -> db.redisClient.del('presets_hash', (err) ->
throw err if err throw err if err
) )
@@ -172,7 +171,41 @@ listener = (request, response) ->
response.end() response.end()
) )
else if request.method == 'DELETE' else if request.method == 'DELETE'
db.scriptsCollection.remove({'_id' : scriptName}, (err) -> db.provisionsCollection.remove({'_id' : provisionName}, (err) ->
db.redisClient.del('presets_hash', (err) ->
throw err if err
)
if err
response.writeHead(500)
response.end(errorToString(err))
return
response.writeHead(200)
response.end()
)
else
response.writeHead 405, {'Allow': 'PUT, DELETE'}
response.end('405 Method Not Allowed')
else if VIRTUAL_PARAMETERS_REGEX.test(urlParts.pathname)
virtualParameterName = querystring.unescape(VIRTUAL_PARAMETERS_REGEX.exec(urlParts.pathname)[1])
if request.method == 'PUT'
object = {
_id: virtualParameterName
script: body.toString()
}
db.virtualParametersCollection.save(object, (err) ->
db.redisClient.del('presets_hash', (err) ->
throw err if err
)
if err
response.writeHead(500)
response.end(errorToString(err))
return
response.writeHead(200)
response.end()
)
else if request.method == 'DELETE'
db.virtualParametersCollection.remove({'_id' : virtualParameterName}, (err) ->
db.redisClient.del('presets_hash', (err) -> db.redisClient.del('presets_hash', (err) ->
throw err if err throw err if err
) )
+9 -12
View File
@@ -38,9 +38,10 @@ sandbox.context.declare = (decs) ->
if ++ sandbox.revision > sandbox.maxRevision + 1 if ++ sandbox.revision > sandbox.maxRevision + 1
throw new Error('Declare function should not be called from within a try/catch block') throw new Error('Declare function should not be called from within a try/catch block')
sandbox.declarations = decs
if sandbox.revision == sandbox.maxRevision + 1 if sandbox.revision == sandbox.maxRevision + 1
for k, v of decs
sandbox.declarations.push([common.parsePath(k)].concat(toIndexDeclaration(v)))
throw new Declare() throw new Declare()
res = {} res = {}
@@ -98,24 +99,20 @@ toIndexDeclaration = (keyDeclaraiton) ->
return indexDeclaraiton return indexDeclaraiton
run = (script, args, deviceData, revision, declarations) -> run = (script, args, deviceData, startRevision, maxRevision) ->
sandbox.deviceData = deviceData sandbox.deviceData = deviceData
sandbox.args = args sandbox.args = args
sandbox.maxRevision = revision sandbox.revision = startRevision
sandbox.revision = 0 sandbox.maxRevision = maxRevision
sandbox.declarations = [] sandbox.declarations = []
try try
script.runInNewContext(sandbox.context) ret = script.runInNewContext(sandbox.context)
return {done: true, declarations: sandbox.declarations, returnValue: ret}
catch err catch err
throw err if err not instanceof Declare throw err if err not instanceof Declare
for k, v of sandbox.declarations
declarations.push([common.parsePath(k)].concat(toIndexDeclaration(v)))
return true
for k, v of sandbox.declarations return {done: false, declarations: sandbox.declarations}
declarations.push([common.parsePath(k)].concat(toIndexDeclaration(v)))
return false
exports.run = run exports.run = run
+312 -200
View File
@@ -37,7 +37,9 @@ init = (deviceId, cwmpVersion, timeout) ->
cwmpVersion : cwmpVersion cwmpVersion : cwmpVersion
timeout : timeout timeout : timeout
provisions: [] provisions: []
counter : [0, 0, 0] # rpc, revision, batch revisions: [0]
rpcCount: 0
cycle: 0
} }
return sessionData return sessionData
@@ -67,12 +69,10 @@ loadParameters = (sessionData, toLoad, callback) ->
generateRpcId = (sessionData) -> generateRpcId = (sessionData) ->
if sessionData.counter[0] > 255 or sessionData.counter[1] > 15 or sessionData.counter[2] > 15 if sessionData.rpcCount > 255 or sessionData.cycle > 15 or sessionData.revisions.length > 15
throw new Error('Too many RPCs') throw new Error('Too many RPCs')
return sessionData.counter[2].toString(16).slice(-1) + return sessionData.timestamp.toString(16) + "0#{sessionData.rpcCount.toString(16)}".slice(-2)
sessionData.counter[1].toString(16).slice(-1) +
('0' + sessionData.counter[0].toString(16)).slice(-2)
inform = (sessionData, rpcReq, callback) -> inform = (sessionData, rpcReq, callback) ->
@@ -95,205 +95,249 @@ inform = (sessionData, rpcReq, callback) ->
addProvisions = (sessionData, provisions) -> addProvisions = (sessionData, provisions) ->
sessionData.provisions = sessionData.provisions.concat(provisions) sessionData.provisions = sessionData.provisions.concat(provisions)
if sessionData.counter[1] > 0 if sessionData.revisions.length > 1 or sessionData.revisions[0] > 0
sessionData.counter[2] += 1 sessionData.cycle += 1
sessionData.counter[1] = 0 sessionData.rpcCount = 0
sessionData.counter[0] = 0 sessionData.revisions = [0]
device.collapse(sessionData.deviceData, 1)
clearProvisions = (sessionData) -> clearProvisions = (sessionData) ->
sessionData.provisions = [] sessionData.provisions = []
if sessionData.counter[1] > 0 if sessionData.revisions.length > 1 or sessionData.revisions[0] > 0
sessionData.counter[2] += 1 sessionData.cycle += 1
sessionData.counter[1] = 0 sessionData.rpcCount = 0
sessionData.counter[0] = 0 sessionData.revisions = [0]
device.collapse(sessionData.deviceData, 1)
generateRpcRequest = (sessionData, declarations, callback) -> generateRpcRequest = (sessionData) ->
test = (cb) -> res = device.traverse(sessionData.deviceData, null, null, (path, declaration, base, current, descendantTimestamps, children) ->
toLoad = [] if path[0] == 'Tags'
res = device.traverse(sessionData.deviceData, null, null, (path, declaration, base, current, descendantTimestamps, children) -> if path.length == 2 and path[1]? and declaration[7]?
if declaration[7][0] != current[7][0]
if path[0] == 'Tags' device.set(sessionData.deviceData, path, 1, [sessionData.timestamp, 1, null, null, null, null, sessionData.timestamp, declaration[7]])
if path.length == 2 and path[1]? and declaration[7]?
if declaration[7][0] != current[7][0]
device.set(sessionData.deviceData, path, 1, [sessionData.timestamp, 1, null, null, null, null, sessionData.timestamp, declaration[7]])
return
r = {}
gpn = false
descendantFlags = []
for i in [0...descendantTimestamps.length] by 1
if descendantTimestamps[i][2]?
if not descendantTimestamps[i][4]?
toLoad.push(descendantTimestamps[i][0])
continue
else if descendantTimestamps[i][2] > descendantTimestamps[i][4]
descendantFlags[i] = 1
descendantFlags[i] |= 2 if descendantTimestamps[i][0].length > path.length + 1
else if not descendantTimestamps[i][4]?
continue
for j in [0...i] by 1 when descendantTimestamps[j][4]? or descendantFlags[j]
overlap = common.pathOverlap(descendantTimestamps[j][0], descendantTimestamps[i][0], path.length)
if overlap & 1 and descendantTimestamps[j][4]?
if descendantFlags[i] & 2 and descendantTimestamps[j][0].length == path.length + 1
descendantFlags[i] ^= 2
if descendantFlags[i] & 1 and descendantTimestamps[j][4] > descendantTimestamps[i][2]
descendantFlags[i] ^= 1
if overlap & 2 and descendantTimestamps[i][4]?
if descendantFlags[j] & 2 and descendantTimestamps[i][0].length == path.length + 1
descendantFlags[j] ^= 2
if descendantFlags[j] & 1 and descendantTimestamps[i][4] > descendantTimestamps[j][2]
descendantFlags[j] ^= 1
for d, j in descendantFlags
if d & 1
gpn = true
if d & 2
toLoad.push(path.concat(descendantTimestamps[j][0][path.length] ? null))
if declaration.length
if declaration[0]?
if not current[0]?
toLoad.push(path)
else if declaration[0] > current[0]
r.exist = true
if declaration[2]?
if not current[2]?
toLoad.push(path)
else if declaration[2] > current[2]
r.writable = true
r.exist = false
if declaration[4]?
if not current[4]?
toLoad.push(path)
else if declaration[4] > current[4]
r.object = true
r.exist = false
if declaration[6]?
if not current[6]?
toLoad.push(path) if not current[3]
else if declaration[6] > current[6]
if current[7]? or current[3] == 0
r.gpv ?= []
r.gpv.push(path)
r.object = false
r.exist = false
gpn = false
else if not current[3]?
r.object = true
if declaration[7]? and current[7]?
if not current[5]?
toLoad.push(path)
else if current[4] == 0
r.writable = true
else if current[5]
if declaration[7][0] != current[7][0] or declaration[7][1] != current[7][1]
r.spv = [[path, declaration[7][0], declaration[7][1]]]
r.exist = false
gpn = false
for k, v of children
if v.writable or v.object or v.exist
gpn = true
if v.gpn
r.gpn = (r.gpn ? []).concat(v.gpn)
r.exist = false
if v.gpv
r.gpv = (r.gpv ? []).concat(v.gpv)
r.exist = false
if v.spv
r.spv = (r.spv ? []).concat(v.spv)
r.exist = false
if gpn
r.exist = false
if r.gpn?
r.gpn = [path].concat(r.gpn)
else
r.gpn = [path]
return r
)
if toLoad.length
loadParameters(sessionData, toLoad, (err) ->
return callback(err) if err
test(cb)
)
return return
if res.gpn? r = {}
rpcReq = { gpn = false
type: 'GetParameterNames'
parameterPath: res.gpn[0].join('.')
nextLevel: true
}
return callback(null, rpcReq)
if res.gpv? descendantRefresh = []
rpcReq = { for i in [0...descendantTimestamps.length] by 1 when descendantTimestamps[i][4]? or descendantTimestamps[i][2]?
type: 'GetParameterValues' if descendantTimestamps[i][2] > descendantTimestamps[i][4]
parameterNames: (p.join('.') for p in res.gpv.slice(0, config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId))) if current[3] == 1
} descendantRefresh[i] = true
return callback(null, rpcReq) else
r.object = true if current[2] < descendantTimestamps[i][2]
if res.spv? for j in [0...i] by 1 when descendantTimestamps[j][4]? or descendantRefresh[j]
rpcReq = { overlap = common.pathOverlap(descendantTimestamps[j][0], descendantTimestamps[i][0], path.length)
type: 'SetParameterValues'
parameterList: ([p[0].join('.'), p[1], p[2]] for p in res.spv.slice(0, config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId)))
}
return callback(null, rpcReq)
return cb() if overlap & 1 and descendantTimestamps[j][4]?
if descendantRefresh[i] and descendantTimestamps[j][4] > descendantTimestamps[i][2]
descendantRefresh[i] = false
for declaration in declarations if overlap & 2 and descendantTimestamps[i][4]?
pre = device.getPrerequisiteDeclarations(declaration) if descendantRefresh[j] and descendantTimestamps[i][4] > descendantTimestamps[j][2]
for d in pre descendantRefresh[j] = false
device.declare(sessionData.deviceData, d[0], d.slice(1), sessionData.timestamp)
test(() -> for d in descendantRefresh when d
for declaration in declarations gpn = true
params = device.getAll(sessionData.deviceData, declaration[0], null) break
if declaration[0][0] == 'Tags' and declaration[0].length == 2
if not declaration[0][1]?
continue if declaration[8][0]
else if params.length == 0
device.set(sessionData.deviceData, declaration[0], 1, [sessionData.timestamp, 1, null, null, null, null, sessionData.timestamp, [false, 'xsd:boolean']])
device.declare(sessionData.deviceData, declaration[0], declaration.slice(1), sessionData.timestamp)
continue
for p in params if declaration.length
device.declare(sessionData.deviceData, p[0], declaration.slice(1), sessionData.timestamp) if declaration[0] > current[0]
r.exist = true
test(() -> if declaration[2] > current[2]
return callback() r.writable = true
) r.exist = false
if declaration[4] > current[4]
r.object = true
r.exist = false
if declaration[6] > current[6]
if current[7]? or current[3] == 0
r.gpv ?= []
r.gpv.push(path)
r.object = false
r.exist = false
gpn = false
else if not current[3]?
r.object = true
if declaration[7]? and current[7]?
if current[4] == 0
r.writable = true
else if current[5]
if declaration[7][0] != current[7][0] or declaration[7][1] != current[7][1]
r.spv = [[path, declaration[7][0], declaration[7][1]]]
r.exist = false
gpn = false
for k, v of children
if v.writable or v.object or v.exist
gpn = true
if v.gpn
r.gpn = (r.gpn ? []).concat(v.gpn)
r.exist = false
if v.gpv
r.gpv = (r.gpv ? []).concat(v.gpv)
r.exist = false
if v.spv
r.spv = (r.spv ? []).concat(v.spv)
r.exist = false
if gpn
r.exist = false
if r.gpn?
r.gpn = [path].concat(r.gpn)
else
r.gpn = [path]
return r
) )
if res.gpn?
rpcReq = {
type: 'GetParameterNames'
parameterPath: res.gpn[0].join('.')
nextLevel: true
}
return rpcReq
rpcRequest = (sessionData, allDeclarations, callback) -> if res.gpv?
allDeclarations = allDeclarations?.slice() ? [] rpcReq = {
type: 'GetParameterValues'
parameterNames: (p.join('.') for p in res.gpv.slice(0, config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId)))
}
return rpcReq
nextRevision = false if res.spv?
rpcReq = {
type: 'SetParameterValues'
parameterList: ([p[0].join('.'), p[1], p[2]] for p in res.spv.slice(0, config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId)))
}
return rpcReq
cache.getScripts((err, presetsHash, scripts) -> return null
loadDeclarations = (sessionData, virtualParameters, callback) ->
toLoad = []
device.traverse(sessionData.deviceData, null, null, (path, declaration, base, current, descendantTimestamps, children) ->
descendantLoad = []
for i in [0...descendantTimestamps.length] by 1
if descendantTimestamps[i][2]?
if not descendantTimestamps[i][4]?
toLoad.push(descendantTimestamps[i][0])
continue
else if descendantTimestamps[i][2] > descendantTimestamps[i][4]
descendantLoad[i] = true if descendantTimestamps[i][0].length > path.length + 1
else if not descendantTimestamps[i][4]?
continue
for j in [0...i] by 1 when descendantTimestamps[j][4]? or descendantFlags[j]
overlap = common.pathOverlap(descendantTimestamps[j][0], descendantTimestamps[i][0], path.length)
if overlap & 1 and descendantTimestamps[j][4]?
if descendantLoad[i] and descendantTimestamps[j][0].length == path.length + 1
descendantLoad[i] = false
if overlap & 2 and descendantTimestamps[i][4]?
if descendantLoad[j] and descendantTimestamps[i][0].length == path.length + 1
descendantLoad[j] = false
for d, j in descendantLoad when d
toLoad.push(path.concat(descendantTimestamps[j][0][path.length] ? null))
for i in [0...declaration.length] by 2
if declaration[i]? and not current[i]?
toLoad.push(path)
break
)
# Virtual parameters
for p in toLoad
if not p[0]? or p[0] == 'VirtualParameters'
if p.length == 1
device.set(sessionData.deviceData, ['VirtualParameters'], 1, [sessionData.timestamp, 1, sessionData.timestamp, 1, sessionData.timestamp, 0])
else if p.length == 2
if not p[1]?
for k, v of virtualParameters
device.set(sessionData.deviceData, ['VirtualParameters', k], 1, [sessionData.timestamp, 1, sessionData.timestamp, 0])
else if virtualParameters[p[1]]?
device.set(sessionData.deviceData, ['VirtualParameters', p[1]], 1, [sessionData.timestamp, 1, sessionData.timestamp, 0])
return loadParameters(sessionData, toLoad, callback)
extractVirtualParameterDeclarations = (sessionData) ->
virtualParameterDeclarations = []
res = device.traverse(sessionData.deviceData, ['VirtualParameters', null], null, (path, declaration, base, current, descendantTimestamps, children) ->
if path.length == 2 and declaration?.length
d = [path]
for i in [0...declaration.length] by 2
if declaration[i]? and not (declaration[i] <= current[i])
d[i + 1] = declaration[i]
if declaration[i + 1]?
if i != 6 or not current[i + 1]?
d[i + 2] = declaration[i + 1]
else if declaration[i + 1][0] != current[i + 1][0] and
(declaration[i + 1][0] != current[i + 1][0] or not declaration[i + 1][1]?)
d[i + 2] = declaration[i + 1]
if d.length > 1
virtualParameterDeclarations.push(d)
)
device.clearDeclarations(sessionData.deviceData, ['VirtualParameters'])
return virtualParameterDeclarations
commitVirtualParameter = (sessionData, parameterDeclaration, revision, update) ->
v = []
if update.writable?
if update.writable[0] <= 0
v[4] = now + update.writable[0]
else
v[4] = Math.min(sessionData.timestamp, update.writable[0])
if parameterDeclaration[5]?
v[4] = Math.max(parameterDeclaration[5], v[4])
v[5] = Boolean(JSON.parse(update.writable[1]))
else if parameterDeclaration[5]? or parameterDeclaration[6]?
throw new Error('Virtual parameter must provide declared attributes')
if update.value?
if update.value[0] <= 0
v[6] = sessionData.timestamp + update.value[0]
else
v[6] = Math.min(now, update.value[0])
if parameterDeclaration[7]?
v[6] = Math.max(parameterDeclaration[7], v[6])
v[7] = device.sanitizeParameterValue(update.value[1])
else if parameterDeclaration[7]? or parameterDeclaration[8]?
throw new Error('Virtual parameter must provide declared attributes')
device.set(sessionData.deviceData, parameterDeclaration[0], revision, v)
rpcRequest = (sessionData, declarations, callback) ->
allDeclarations = declarations?.slice() ? []
cache.getProvisionsAndVirtualParameters((err, presetsHash, provisions, virtualParameters) ->
return callback(err) if err or presetsHash != sessionData.presetsHash return callback(err) if err or presetsHash != sessionData.presetsHash
done = true
for provision in sessionData.provisions for provision in sessionData.provisions
if not scripts[provision[0]]? if not provisions[provision[0]]?
switch provision[0] switch provision[0]
when 'refresh' when 'refresh'
path = common.parsePath(provision[1]) path = common.parsePath(provision[1])
@@ -306,63 +350,131 @@ rpcRequest = (sessionData, allDeclarations, callback) ->
allDeclarations.push([[['Tags', provision[1]], null, null, null, null, null, null, null, [provision[2], 'xsd:boolean']]]) allDeclarations.push([[['Tags', provision[1]], null, null, null, null, null, null, null, [provision[2], 'xsd:boolean']]])
continue continue
nextRevision ||= sandbox.run(scripts[provision[0]].compiled, provision.slice(1), sessionData.deviceData, sessionData.counter[1], allDeclarations) ret = sandbox.run(provisions[provision[0]].script, provision.slice(1), sessionData.deviceData, 0, sessionData.revisions[0] >> 1)
done &&= ret.done
allDeclarations = allDeclarations.concat(ret.declarations)
generateRpcRequest(sessionData, allDeclarations, (err, rpcReq) -> if done and (not allDeclarations?.length or sessionData.revisions[0] > 1)
sessionData.rpcRequest = rpcReq return callback()
return callback(err) if err
if not rpcReq? doVirtualParameters = (iter, virtualParameterDeclarations, cb) ->
if not nextRevision if not virtualParameterDeclarations?
return callback() virtualParameterDeclarations = extractVirtualParameterDeclarations(sessionData)
sessionData.counter[0] = 0 if not virtualParameterDeclarations?.length
sessionData.counter[1] += 1 return cb(false)
return rpcRequest(sessionData, null, callback)
callback(err, generateRpcId(sessionData), rpcReq) sessionData.revisions[iter] ?= 0
decs = []
virtualParameterUpdates = []
for vpd, i in virtualParameterDeclarations
vpName = vpd[0][1]
continue if not virtualParameters[vpName]?
lastRevision = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1))
firstRevision = lastRevision - (sessionData.revisions[sessionData.revisions.length - 1] >> 1)
ret = sandbox.run(virtualParameters[vpName].script, [vpd.slice(1)], sessionData.deviceData, firstRevision, lastRevision)
decs = decs.concat(ret.declarations)
if ret.done
virtualParameterUpdates.push(ret.returnValue)
if virtualParameterUpdates.length == virtualParameterDeclarations.length
rev = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1))
for vpd, i in virtualParameterDeclarations
commitVirtualParameter(sessionData, vpd, rev + 1, virtualParameterUpdates[i])
if sessionData.revisions[iter] == 0
return cb(false)
sessionData.revisions.length = iter
rev = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1))
device.collapse(sessionData.deviceData, rev + 1)
return cb(true)
applyDeclarations(iter, decs, () ->
return doVirtualParameters(iter, virtualParameterDeclarations, cb)
)
applyDeclarations = (iter, decs, cb) ->
if sessionData.revisions[iter] % 2 == 0
for declaration in decs
for d in device.getPrerequisiteDeclarations(declaration)
device.declare(sessionData.deviceData, d[0], d.slice(1), sessionData.timestamp)
else
for declaration in decs
params = device.getAll(sessionData.deviceData, declaration[0], sessionData.revisions[iter] >> 1)
if declaration[0][0] == 'Tags' and declaration[0].length == 2
if not declaration[0][1]?
continue if declaration[8][0]
else if params.length == 0
device.set(sessionData.deviceData, declaration[0], 1, [sessionData.timestamp, 1, null, null, null, null, sessionData.timestamp, [false, 'xsd:boolean']])
device.declare(sessionData.deviceData, declaration[0], declaration.slice(1), sessionData.timestamp)
continue
for p in params
device.declare(sessionData.deviceData, p[0], declaration.slice(1), sessionData.timestamp)
nextRpc = (applied) ->
if not applied
if (rpcReq = generateRpcRequest(sessionData))?
sessionData.rpcRequest = rpcReq
return callback(null, generateRpcId(sessionData), rpcReq)
if ++ sessionData.revisions[iter] % 2 == 0
return cb()
else
return applyDeclarations(iter, decs, cb)
if sessionData.revisions.length > iter + 1
doVirtualParameters(iter + 1, null, nextRpc)
else
loadDeclarations(sessionData, virtualParameters, (err) ->
return callback(err) if err
doVirtualParameters(iter + 1, null, nextRpc)
)
applyDeclarations(0, allDeclarations, () ->
return rpcRequest(sessionData, declarations, callback)
) )
) )
return
rpcResponse = (sessionData, id, rpcRes, callback) -> rpcResponse = (sessionData, id, rpcRes, callback) ->
if id != generateRpcId(sessionData) if id != generateRpcId(sessionData)
return callback(new Error('Request ID not recognized')) return callback(new Error('Request ID not recognized'))
++ sessionData.counter[0] ++ sessionData.rpcCount
rpcReq = sessionData.rpcRequest rpcReq = sessionData.rpcRequest
sessionData.rpcRequest = null sessionData.rpcRequest = null
timestamp = sessionData.timestamp timestamp = sessionData.timestamp
revision = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1)) + 1
switch rpcRes.type switch rpcRes.type
when 'GetParameterValuesResponse' when 'GetParameterValuesResponse'
return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'GetParameterValues' return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'GetParameterValues'
for p in rpcRes.parameterList for p in rpcRes.parameterList
device.set(sessionData.deviceData, common.parsePath(p[0]), 1, [timestamp, 1, timestamp, 0, null, null, timestamp, p.slice(1)]) device.set(sessionData.deviceData, common.parsePath(p[0]), revision, [timestamp, 1, timestamp, 0, null, null, timestamp, p.slice(1)])
when 'GetParameterNamesResponse' when 'GetParameterNamesResponse'
return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'GetParameterNames' return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'GetParameterNames'
device.set(sessionData.deviceData, common.parsePath(rpcReq.parameterPath).concat(null), 1, timestamp) device.set(sessionData.deviceData, common.parsePath(rpcReq.parameterPath).concat(null), revision, timestamp)
for p in rpcRes.parameterList for p in rpcRes.parameterList
if common.endsWith(p[0], '.') if common.endsWith(p[0], '.')
path = common.parsePath(p[0][0...-1]) path = common.parsePath(p[0][0...-1])
if not rpcReq.nextLevel if not rpcReq.nextLevel
device.set(sessionData.deviceData, path.conact(null), 1, timestamp) device.set(sessionData.deviceData, path.conact(null), revision, timestamp)
device.set(sessionData.deviceData, path, 1, [timestamp, 1, timestamp, 1, timestamp, if p[1] then 1 else 0]) device.set(sessionData.deviceData, path, revision, [timestamp, 1, timestamp, 1, timestamp, if p[1] then 1 else 0])
else else
device.set(sessionData.deviceData, common.parsePath(p[0]), 1, [timestamp, 1, timestamp, 0, timestamp, if p[1] then 1 else 0]) device.set(sessionData.deviceData, common.parsePath(p[0]), revision, [timestamp, 1, timestamp, 0, timestamp, if p[1] then 1 else 0])
when 'SetParameterValuesResponse' when 'SetParameterValuesResponse'
return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'SetParameterValues' return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'SetParameterValues'
for p in rpcReq.parameterList for p in rpcReq.parameterList
device.set(sessionData.deviceData, common.parsePath(p[0]), 1, [timestamp, 1, timestamp, 0, timestamp, 1, timestamp, p.slice(1)]) device.set(sessionData.deviceData, common.parsePath(p[0]), revision, [timestamp, 1, timestamp, 0, timestamp, 1, timestamp, p.slice(1)])
else else
return callback(new Error('Response type not recognized')) return callback(new Error('Response type not recognized'))