diff --git a/lib/cache.coffee b/lib/cache.coffee index 3d77608..205ffe4 100644 --- a/lib/cache.coffee +++ b/lib/cache.coffee @@ -25,7 +25,8 @@ REFRESH = 3000 nextRefresh = 0 hash = 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' @@ -55,18 +56,22 @@ lock = (lockName, ttl, callback) -> computeHash = () -> - # MD5 hash for presets and scripts for detecting changes + # MD5 hash for presets, provisions, virtual parameters for detecting changes h = crypto.createHash('md5') for p in presets h.update(JSON.stringify(p.name)) h.update(JSON.stringify(p.precondition)) h.update(JSON.stringify(p.provisions)) - for s in Object.keys(scripts).sort() - h.update(JSON.stringify(s)) - h.update(JSON.stringify(scripts[s].source)) + keys = Object.keys(provisions).sort() + h.update(JSON.stringify(keys)) + 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) -> @@ -79,7 +84,7 @@ refresh = (callback) -> return callback() lock('presets_hash_lock', 3000, (err, unlockOrExtend) -> - counter = 2 + counter = 3 db.presetsCollection.find().toArray((err, res) -> if err callback(err) if -- counter >= 0 @@ -98,25 +103,25 @@ refresh = (callback) -> for preset in res precondition = JSON.parse(preset.precondition) - provisions = preset.provisions or [] + _provisions = preset.provisions or [] # Generate provisions from the old configuration format for c in preset.configurations switch c.type when 'age' - provisions.push(['refresh', c.name, c.age * -1000]) + _provisions.push(['refresh', c.name, c.age * -1000]) when 'value' - provisions.push(['value', c.name, c.value]) + _provisions.push(['value', c.name, c.value]) when 'add_tag' - provisions.push(['tag', c.tag, true]) + _provisions.push(['tag', c.tag, true]) when 'delete_tag' - provisions.push(['tag', c.tag, false]) - when 'script' - provisions.push([c.name].concat(c.args or [])) + _provisions.push(['tag', c.tag, false]) + when 'provision' + _provisions.push([c.name].concat(c.args or [])) else 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 computeHash() @@ -127,18 +132,38 @@ refresh = (callback) -> ) ) - db.scriptsCollection.find().toArray((err, res) -> + db.provisionsCollection.find().toArray((err, res) -> if err callback(err) if -- counter >= 0 counter = 0 return - scripts = {} + provisions = {} for r in res - scripts[r._id] = {} - scripts[r._id].source = r.source - scripts[r._id].md5 = r.md5 - scripts[r._id].compiled = vm.Script(r.source, {filename: r._id, timeout: 50}) + provisions[r._id] = {} + provisions[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex') + provisions[r._id].script = vm.Script(r.script, {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 computeHash() @@ -162,14 +187,32 @@ getPresets = (callback) -> ) -getScripts = (callback) -> +getProvisions = (callback) -> if Date.now() < nextRefresh - return callback(null, hash, scripts) + return callback(null, hash, provisions) 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.getScripts = getScripts +exports.getProvisions = getProvisions +exports.getProvisionsAndVirtualParameters = getProvisionsAndVirtualParameters diff --git a/lib/db.coffee b/lib/db.coffee index f7adf69..0b4b982 100644 --- a/lib/db.coffee +++ b/lib/db.coffee @@ -26,11 +26,12 @@ tasksCollection = null devicesCollection = null presetsCollection = null objectsCollection = null -scriptsCollection = null +provisionsCollection = null +virtualParametersCollection = null connect = (callback) -> - callbackCounter = 7 + callbackCounter = 8 mongodb.MongoClient.connect(config.get('MONGODB_CONNECTION_URL'), {db:{w:1},server:{autoReconnect:true}}, (err, db) -> return callback(err) if err exports.mongoDb = db @@ -72,8 +73,15 @@ connect = (callback) -> return callback(err) ) - db.collection('scripts', (err, collection) -> - exports.scriptsCollection = scriptsCollection = collection + db.collection('provisions', (err, 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 callbackCounter = 0 return callback(err) diff --git a/lib/device.coffee b/lib/device.coffee index 557fa8c..28e1040 100644 --- a/lib/device.coffee +++ b/lib/device.coffee @@ -98,7 +98,7 @@ init = () -> declare = (deviceData, path, declaration, now) -> 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 break @@ -137,7 +137,7 @@ declare = (deviceData, path, declaration, now) -> set = (deviceData, path, revision, properties) -> ++ revision # revisions start at layer 1 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 break @@ -343,6 +343,45 @@ traverse = (deviceData, pattern, revision, callback) -> 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) -> dec = [declarations[0], 1] for i in [1...declarations.length] by 2 @@ -368,3 +407,6 @@ exports.getAll = getAll exports.declare = declare exports.set = set exports.traverse = traverse +exports.collapse = collapse +exports.clearDeclarations = clearDeclarations +exports.sanitizeParameterValue = sanitizeParameterValue diff --git a/lib/nbi.coffee b/lib/nbi.coffee index 2f19cd5..ad536f3 100644 --- a/lib/nbi.coffee +++ b/lib/nbi.coffee @@ -41,7 +41,6 @@ url = require 'url' mongodb = require 'mongodb' querystring = require 'querystring' -crypto = require 'crypto' config = require './config' common = require './common' @@ -59,7 +58,8 @@ FILES_REGEX = /^\/files\/([a-zA-Z0-9\%\!\*\'\(\)\;\:\@\&\=\+\$\,\?\#\[\]\-\_\.\~ PING_REGEX = /^\/ping\/([a-zA-Z0-9\-\_\.]+)\/?$/ QUERY_REGEX = /^\/([a-zA-Z0-9_]+s)\/?$/ 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) -> @@ -151,16 +151,15 @@ listener = (request, response) -> else response.writeHead 405, {'Allow': 'PUT, DELETE'} response.end('405 Method Not Allowed') - else if SCRIPTS_REGEX.test(urlParts.pathname) - scriptName = querystring.unescape(SCRIPTS_REGEX.exec(urlParts.pathname)[1]) + else if PROVISIONS_REGEX.test(urlParts.pathname) + provisionName = querystring.unescape(PROVISIONS_REGEX.exec(urlParts.pathname)[1]) if request.method == 'PUT' object = { - _id: scriptName - source: body.toString() - md5: crypto.createHash('md5').update(body).digest('hex') + _id: provisionName + script: body.toString() } - db.scriptsCollection.save(object, (err) -> + db.provisionsCollection.save(object, (err) -> db.redisClient.del('presets_hash', (err) -> throw err if err ) @@ -172,7 +171,41 @@ listener = (request, response) -> response.end() ) 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) -> throw err if err ) diff --git a/lib/sandbox.coffee b/lib/sandbox.coffee index cd3b6a9..b736949 100644 --- a/lib/sandbox.coffee +++ b/lib/sandbox.coffee @@ -38,9 +38,10 @@ sandbox.context.declare = (decs) -> if ++ sandbox.revision > sandbox.maxRevision + 1 throw new Error('Declare function should not be called from within a try/catch block') - sandbox.declarations = decs - if sandbox.revision == sandbox.maxRevision + 1 + for k, v of decs + sandbox.declarations.push([common.parsePath(k)].concat(toIndexDeclaration(v))) + throw new Declare() res = {} @@ -98,24 +99,20 @@ toIndexDeclaration = (keyDeclaraiton) -> return indexDeclaraiton -run = (script, args, deviceData, revision, declarations) -> +run = (script, args, deviceData, startRevision, maxRevision) -> sandbox.deviceData = deviceData sandbox.args = args - sandbox.maxRevision = revision - sandbox.revision = 0 + sandbox.revision = startRevision + sandbox.maxRevision = maxRevision sandbox.declarations = [] try - script.runInNewContext(sandbox.context) + ret = script.runInNewContext(sandbox.context) + return {done: true, declarations: sandbox.declarations, returnValue: ret} catch err 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 - declarations.push([common.parsePath(k)].concat(toIndexDeclaration(v))) - return false + return {done: false, declarations: sandbox.declarations} exports.run = run diff --git a/lib/session.coffee b/lib/session.coffee index 303a945..ceb27ed 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -37,7 +37,9 @@ init = (deviceId, cwmpVersion, timeout) -> cwmpVersion : cwmpVersion timeout : timeout provisions: [] - counter : [0, 0, 0] # rpc, revision, batch + revisions: [0] + rpcCount: 0 + cycle: 0 } return sessionData @@ -67,12 +69,10 @@ loadParameters = (sessionData, toLoad, callback) -> 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') - return sessionData.counter[2].toString(16).slice(-1) + - sessionData.counter[1].toString(16).slice(-1) + - ('0' + sessionData.counter[0].toString(16)).slice(-2) + return sessionData.timestamp.toString(16) + "0#{sessionData.rpcCount.toString(16)}".slice(-2) inform = (sessionData, rpcReq, callback) -> @@ -95,205 +95,249 @@ inform = (sessionData, rpcReq, callback) -> addProvisions = (sessionData, provisions) -> sessionData.provisions = sessionData.provisions.concat(provisions) - if sessionData.counter[1] > 0 - sessionData.counter[2] += 1 - sessionData.counter[1] = 0 - sessionData.counter[0] = 0 + if sessionData.revisions.length > 1 or sessionData.revisions[0] > 0 + sessionData.cycle += 1 + sessionData.rpcCount = 0 + sessionData.revisions = [0] + device.collapse(sessionData.deviceData, 1) clearProvisions = (sessionData) -> sessionData.provisions = [] - if sessionData.counter[1] > 0 - sessionData.counter[2] += 1 - sessionData.counter[1] = 0 - sessionData.counter[0] = 0 + if sessionData.revisions.length > 1 or sessionData.revisions[0] > 0 + sessionData.cycle += 1 + sessionData.rpcCount = 0 + sessionData.revisions = [0] + device.collapse(sessionData.deviceData, 1) -generateRpcRequest = (sessionData, declarations, callback) -> - test = (cb) -> - toLoad = [] - res = device.traverse(sessionData.deviceData, null, null, (path, declaration, base, current, descendantTimestamps, children) -> - - if path[0] == 'Tags' - 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) - ) +generateRpcRequest = (sessionData) -> + res = device.traverse(sessionData.deviceData, null, null, (path, declaration, base, current, descendantTimestamps, children) -> + if path[0] == 'Tags' + 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 - if res.gpn? - rpcReq = { - type: 'GetParameterNames' - parameterPath: res.gpn[0].join('.') - nextLevel: true - } - return callback(null, rpcReq) + r = {} + gpn = false - if res.gpv? - rpcReq = { - type: 'GetParameterValues' - parameterNames: (p.join('.') for p in res.gpv.slice(0, config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId))) - } - return callback(null, rpcReq) + descendantRefresh = [] + for i in [0...descendantTimestamps.length] by 1 when descendantTimestamps[i][4]? or descendantTimestamps[i][2]? + if descendantTimestamps[i][2] > descendantTimestamps[i][4] + if current[3] == 1 + descendantRefresh[i] = true + else + r.object = true if current[2] < descendantTimestamps[i][2] - 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 callback(null, rpcReq) + for j in [0...i] by 1 when descendantTimestamps[j][4]? or descendantRefresh[j] + overlap = common.pathOverlap(descendantTimestamps[j][0], descendantTimestamps[i][0], path.length) - 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 - pre = device.getPrerequisiteDeclarations(declaration) - for d in pre - device.declare(sessionData.deviceData, d[0], d.slice(1), sessionData.timestamp) + if overlap & 2 and descendantTimestamps[i][4]? + if descendantRefresh[j] and descendantTimestamps[i][4] > descendantTimestamps[j][2] + descendantRefresh[j] = false - test(() -> - for declaration in declarations - params = device.getAll(sessionData.deviceData, declaration[0], null) - 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 d in descendantRefresh when d + gpn = true + break - for p in params - device.declare(sessionData.deviceData, p[0], declaration.slice(1), sessionData.timestamp) + if declaration.length + if declaration[0] > current[0] + r.exist = true - test(() -> - return callback() - ) + if declaration[2] > current[2] + 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) -> - allDeclarations = allDeclarations?.slice() ? [] + if res.gpv? + 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 + done = true for provision in sessionData.provisions - if not scripts[provision[0]]? + if not provisions[provision[0]]? switch provision[0] when 'refresh' 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']]]) 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) -> - sessionData.rpcRequest = rpcReq - return callback(err) if err + if done and (not allDeclarations?.length or sessionData.revisions[0] > 1) + return callback() - if not rpcReq? - if not nextRevision - return callback() - sessionData.counter[0] = 0 - sessionData.counter[1] += 1 - return rpcRequest(sessionData, null, callback) + doVirtualParameters = (iter, virtualParameterDeclarations, cb) -> + if not virtualParameterDeclarations? + virtualParameterDeclarations = extractVirtualParameterDeclarations(sessionData) + if not virtualParameterDeclarations?.length + return cb(false) - 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) -> if id != generateRpcId(sessionData) return callback(new Error('Request ID not recognized')) - ++ sessionData.counter[0] + ++ sessionData.rpcCount rpcReq = sessionData.rpcRequest sessionData.rpcRequest = null timestamp = sessionData.timestamp + revision = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1)) + 1 switch rpcRes.type when 'GetParameterValuesResponse' return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'GetParameterValues' 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' 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 if common.endsWith(p[0], '.') path = common.parsePath(p[0][0...-1]) 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 - 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' return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'SetParameterValues' 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 return callback(new Error('Response type not recognized'))