From f9c5272ce035eebf724ba1464e41b988f958aad7 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Thu, 10 Nov 2016 07:07:51 -0800 Subject: [PATCH] A number of fixes in sandbox and declaration processing --- lib/cwmp.coffee | 5 +- lib/extension-wrapper.coffee | 3 +- lib/extensions.coffee | 2 +- lib/sandbox.js | 129 ++++++++++++++++------------------- lib/session.coffee | 87 +++++++++++------------ 5 files changed, 107 insertions(+), 119 deletions(-) diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index ff138da..e1e453a 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -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 diff --git a/lib/extension-wrapper.coffee b/lib/extension-wrapper.coffee index 29cd095..5a5afa1 100644 --- a/lib/extension-wrapper.coffee +++ b/lib/extension-wrapper.coffee @@ -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]) ) ) diff --git a/lib/extensions.coffee b/lib/extensions.coffee index 12db54d..484c1b8 100644 --- a/lib/extensions.coffee +++ b/lib/extensions.coffee @@ -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) -> diff --git a/lib/sandbox.js b/lib/sandbox.js index 1b63b99..4c3042d 100644 --- a/lib/sandbox.js +++ b/lib/sandbox.js @@ -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); } diff --git a/lib/session.coffee b/lib/session.coffee index 2043ba5..dfdc674 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -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'