A number of fixes in sandbox and declaration processing

This commit is contained in:
Zaid Abdulla committed 2016-11-10 07:07:51 -08:00
1 parent 3b987cc0cc
commit f9c5272ce0
5 files changed
+107 -119

No files matched your search

+1 -4
View File
@@ -209,8 +209,6 @@ applyPresets = (currentRequest) ->
session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest) ->
throw err if err
if not rpcRequest?
session.clearProvisions(currentRequest.sessionData)
sendRpcRequest(currentRequest, id, rpcRequest)
)
)
@@ -235,12 +233,11 @@ nextRpc = (currentRequest) ->
currentRequest.sessionData.tasks.splice(j, 1)
break
session.clearProvisions(currentRequest.sessionData)
for task in currentRequest.sessionData.tasks
# Delete if expired
if task.expiry <= currentRequest.sessionData.timestamp
util.log("#{currentRequest.sessionData.deviceId}: Task is expired #{task.name}(#{task._id})")
currentRequest.sessionData.doneTasks ?= []
currentRequest.sessionData.doneTasks.push(String(task._id))
continue
+1 -2
View File
@@ -21,7 +21,6 @@ script = require("#{process.argv[2]}/#{process.argv[3]}")
process.on('message', (message) ->
script[message[1][0]](message[1].slice(1), (err, res) ->
throw err if err
process.send([message[0], res])
process.send([message[0], err, res])
)
)
+1 -1
View File
@@ -30,7 +30,7 @@ queue = {}
messageHandler = (message) ->
func = queue[message[0]]
delete queue[message[0]]
func(null, message[1])
func(message[1], message[2])
run = (args, callback) ->
+59 -70
View File
@@ -132,34 +132,33 @@ function declare(path, timestamps, values) {
let parsedPath = common.parsePath(path);
sandbox.declarations[path] = sandbox.declarations[path] || [parsedPath, 1, {}, null, {}];
let declaration = [parsedPath, 1];
let attrs = {};
for (let attrName in timestamps) {
if (attrName === "path") {
if (timestamps.path > sandbox.declarations[path][1])
sandbox.declarations[path][1] = timestamps.path;
}
else {
attrs[attrName] = 1;
if (!(timestamps[attrName] <= sandbox.declarations[path][2][attrName]))
sandbox.declarations[path][2][attrName] = timestamps[attrName];
}
}
if (values) {
for (let attrName in values) {
if (timestamps)
for (let attrName in timestamps)
if (attrName === "path")
sandbox.declarations[path][3] = values.path;
declaration[1] = timestamps.path;
else {
attrs[attrName] = 1;
if (attrName === 'value' && !Array.isArray(values[attrName]))
sandbox.declarations[path][4].value = [values[attrName]];
else
sandbox.declarations[path][4].value = values[attrName];
declaration[2] = declaration[2] || {};
declaration[2][attrName] = timestamps[attrName];
}
}
}
if (values)
for (let attrName in values)
if (attrName === "path")
declaration[3] = values.path;
else {
attrs[attrName] = 1;
declaration[4] = declaration[4] || {};
if (attrName === "value" && !Array.isArray(values.value))
declaration[4].value = [values.value];
else
declaration[4][attrName] = values[attrName];
}
sandbox.declarations.push(declaration);
return new ParameterWrapper(parsedPath, Object.keys(attrs));
}
@@ -212,7 +211,7 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis
sandbox.revision = startRevision;
sandbox.maxRevision = maxRevision;
sandbox.uncommitted = false;
sandbox.declarations = {};
sandbox.declarations = [];
sandbox.extensions = {};
sandbox.clear = null;
@@ -222,59 +221,49 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis
for (let n in globals)
sandbox.context[n] = globals[n];
let ret, status;
try {
let ret = script.runInContext(sandbox.context, {displayErrors: false});
let declarations = [];
for (let p in sandbox.declarations)
declarations.push(sandbox.declarations[p]);
let counter = 1;
for (let key in sandbox.extensions) {
++ counter;
extensions.run(sandbox.extensions[key], function(err, res) {
sandbox.extensionsCache[key] = res;
if (-- counter === 0)
return callback(null, sandbox.clear, declarations, true, ret);
});
}
if (-- counter === 0)
return callback(null, sandbox.clear, declarations, true, ret);
ret = script.runInContext(sandbox.context, {displayErrors: false});
status = 0;
}
catch (err) {
if (err === COMMIT) {
let declarations = [];
for (let p in sandbox.declarations)
declarations.push(sandbox.declarations[p]);
let counter = 1;
for (let key in sandbox.extensions) {
++ counter;
extensions.run(sandbox.extensions[key], function(err, res) {
sandbox.extensionsCache[key] = res;
if (-- counter === 0)
return callback(null, sandbox.clear, declarations, false);
});
}
if (-- counter === 0)
return callback(null, sandbox.clear, declarations, false);
}
else if (err === EXT) {
let counter = 1;
for (let key in sandbox.extensions) {
++ counter;
extensions.run(sandbox.extensions[key], function(err, res) {
sandbox.extensionsCache[key] = res;
if (-- counter === 0)
return run(script, globals, timestamp, deviceData, extensionsCache, startRevision, maxRevision, callback);
});
}
if (-- counter === 0)
return run(script, globals, timestamp, deviceData, extensionsCache, startRevision, maxRevision, callback);
}
if (err === COMMIT)
status = 1;
else if (err === EXT)
status = 2;
else
return callback(err);
}
let ext = sandbox.extensions;
let decs = sandbox.declarations;
let clr = sandbox.clear;
let extCache = sandbox.extensionsCache;
let args = arguments;
let counter = 1;
for (let key in sandbox.extensions) {
++ counter;
extensions.run(sandbox.extensions[key], function(err, res) {
if (err && counter > 0) {
counter = 0;
return callback(err);
}
extCache[key] = res;
if (-- counter === 0)
if (status === 2)
return run.apply(null, args);
else
return callback(null, clr, decs, status === 0, ret);
});
}
if (-- counter === 0)
if (status === 2)
return run.apply(null, args);
else
return callback(null, clr, decs, status === 0, ret);
}
+45 -42
View File
@@ -157,13 +157,17 @@ inform = (sessionData, rpcReq, callback) ->
addProvisions = (sessionData, channel, provisions) ->
sessionData.declarations = []
sessionData.provisions.length = 1
sessionData.doneProvisions = 0
sessionData.doneDeclarations = 0
if sessionData.revisions[0] > 0
sessionData.deviceData.timestamps.collapse(1)
sessionData.deviceData.attributes.collapse(1)
delete sessionData.syncState
sessionData.declarations = []
sessionData.doneProvisions = 0
sessionData.provisions[0] ?= []
sessionData.revisions[0] ?= 0
sessionData.provisions.length = 1
sessionData.revisions = [0]
sessionData.extensionsCache = {}
for provision, i in provisions
# Remove duplicate provisions
@@ -175,34 +179,25 @@ addProvisions = (sessionData, channel, provisions) ->
sessionData.provisions[0].push(provision)
sessionData.channels.push(channel)
if sessionData.revisions.length > 1 or sessionData.revisions[0] > 0
v.collapse(1) for k, v of sessionData.deviceData.timestamps
v.collapse(1) for k, v of sessionData.deviceData.attributes
sessionData.rpcCount = 0
sessionData.revisions = [0]
sessionData.extensionsCache = {}
clearProvisions = (sessionData) ->
if sessionData.revisions[sessionData.revisions.length - 1] > 0
sessionData.deviceData.timestamps.collapse(1)
sessionData.deviceData.attributes.collapse(1)
delete sessionData.syncState
sessionData.provisions = [[]]
sessionData.provisions = []
sessionData.channels = []
sessionData.declarations = []
sessionData.doneProvisions = 0
sessionData.doneDeclarations = 0
if sessionData.revisions.length > 1 or sessionData.revisions[0] > 0
sessionData.deviceData.timestamps.collapse(1)
sessionData.deviceData.attributes.collapse(1)
sessionData.rpcCount = 0
sessionData.revisions = [0]
sessionData.extensionsCache = {}
sessionData.revisions = []
sessionData.extensionsCache = {}
runProvisions = (sessionData, provisions, startRevision, endRevision, callback) ->
done = true
_extensions = []
_clear = []
allDeclarations = []
allClear = []
counter = 1
for provision in provisions
if not sessionData.cache.provisions[provision[0]]?
@@ -259,22 +254,26 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback)
return
done &&= _done
allDeclarations = allDeclarations.concat(_declarations)
if _declarations
allDeclarations = allDeclarations.concat(_declarations)
if _clear
allClear = allClear.concat(_clear)
if counter == 0
return callback(null, done, allDeclarations, _clear)
return callback(null, done, allDeclarations, allClear)
)
if -- counter == 0
return callback(null, done, allDeclarations, _clear)
return callback(null, done, allDeclarations, allClear)
runVirtualParameters = (sessionData, provisions, startRevision, endRevision, callback) ->
done = true
decs = []
virtualParameterUpdates = []
_extensions = []
_clear = []
allDeclarations = []
allClear = []
counter = 1
for provision in provisions
++ counter
@@ -291,7 +290,12 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
return
done &&= _done
decs = decs.concat(_declarations)
if _declarations
allDeclarations = allDeclarations.concat(_declarations)
if _clear
allClear = allClear.concat(_clear)
if _done
virtualParameterUpdates.push(_returnValue)
@@ -303,7 +307,7 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
toClear = commitVirtualParameter(sessionData, provisions[i], vpu, toClear)
clear(sessionData, toClear, (err) ->
return callback(err, done, decs, _clear)
return callback(err, done, allDeclarations, allClear)
)
)
@@ -314,7 +318,7 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
toClear = commitVirtualParameter(sessionData, provisions[i], vpu, toClear)
clear(sessionData, toClear, (err) ->
return callback(err, done, decs, _clear)
return callback(err, done, allDeclarations, allClear)
)
@@ -354,7 +358,7 @@ runDeclarations = (sessionData, declarations) ->
if (path.alias | path.wildcard) & 1 or path[0] == 'Reboot'
sessionData.deviceData.paths.add(['Reboot'])
if (path.aliasa | path.wildcard) & 1 or path[0] == 'FactoryReset'
if (path.alias | path.wildcard) & 1 or path[0] == 'FactoryReset'
sessionData.deviceData.paths.add(['FactoryReset'])
if path.alias
@@ -411,7 +415,7 @@ runDeclarations = (sessionData, declarations) ->
else
minInstances = maxInstances = declaration[3]
parent = sessionData.deviceData.paths.add(path.slice(0, -1))
parent = common.addPathMeta(path.slice(0, -1))
if Array.isArray(path[path.length - 1])
keys = {}
@@ -421,12 +425,14 @@ runDeclarations = (sessionData, declarations) ->
keys = {}
if ((path.wildcard | path.alias) & ((1 << (path.length - 1)) - 1)) == 0
parent = sessionData.deviceData.paths.add(parent)
unpacked ?= device.unpack(sessionData.device, path)
processInstances(sessionData, parent, unpacked, keys, minInstances, maxInstances)
else
parentsUnpacked = device.unpack(sessionData.deviceData, parent)
for parent in parentsUnpacked
processInstances(sessionData, parent, device.unpack(sessionData.deviceData, parent.concat([path[parent.length]])), keys, minInstances, maxInstances)
parent = sessionData.deviceData.paths.add(parent)
processInstances(sessionData, parent, device.unpack(sessionData.deviceData, common.addPathMeta(parent.concat([path[parent.length]]))), keys, minInstances, maxInstances)
return processDeclarations(sessionData, allDeclareTimestamps, allDeclareAttributeTimestamps, allDeclareAttributeValues)
@@ -437,6 +443,9 @@ rpcRequest = (sessionData, _declarations, callback) ->
if sessionData.declarations.length < sessionData.provisions.length
inception = sessionData.declarations.length
revision = sessionData.revisions[inception] + 1
sessionData.deviceData.timestamps.revision = revision
sessionData.deviceData.attributes.revision = revision
run = if inception == 0 then runProvisions else runVirtualParameters
return run(sessionData, sessionData.provisions[inception], sessionData.revisions[inception - 1] ? 0, sessionData.revisions[inception], (err, done, decs, toClear) ->
@@ -474,7 +483,6 @@ rpcRequest = (sessionData, _declarations, callback) ->
return rpcRequest(sessionData, null, callback)
)
if (sessionData.syncState?.virtualParameterDeclarations?.length or 0) < sessionData.declarations.length
inception = sessionData.syncState?.virtualParameterDeclarations?.length or 0
vpd = runDeclarations(sessionData, sessionData.declarations[inception])
@@ -508,10 +516,6 @@ rpcRequest = (sessionData, _declarations, callback) ->
if not sessionData.syncState?
return callback()
revision = sessionData.revisions[sessionData.revisions.length - 1] + 1
sessionData.deviceData.timestamps.revision = revision
sessionData.deviceData.attributes.revision = revision
inception = sessionData.declarations.length - 1
provisions = generateGetVirtualParameterProvisions(sessionData, sessionData.syncState.virtualParameterDeclarations[inception])
@@ -586,8 +590,7 @@ rpcRequest = (sessionData, _declarations, callback) ->
sessionData.deviceData.timestamps.collapse(rev + 1)
sessionData.deviceData.attributes.collapse(rev + 1)
for k of sessionData.extensionsCache
n = Number(k.split(':', 1)[0])
if n > rev
if rev < Number(k.split(':', 1)[0])
delete sessionData.extensionsCache[k]
sessionData.syncState.virtualParameterDeclarations.length = sessionData.declarations.length
@@ -747,7 +750,7 @@ generateSetRpcRequest = (sessionData) ->
# Delete instance
iter = syncState.instancesToDelete.values()
while instances = iter.next().value
if instance = instances.values().next().value and
if (instance = instances.values().next().value) and
sessionData.deviceData.attributes.has(instance)
return {
type: 'DeleteObject'