diff --git a/lib/common.coffee b/lib/common.coffee index 5bc21f7..d7a6a94 100644 --- a/lib/common.coffee +++ b/lib/common.coffee @@ -235,18 +235,22 @@ parseAlias = (pattern, start, res) -> parsePath = (pattern, start, res) -> path = [] + path.wildcard = 0 + path.alias = 0 i = start ? 0 # Colon separator is needed for parseAlias if i < pattern.length and pattern[i] != ':' while true if pattern[i] == '[' + path.alias |= 1 << path.length i = parseAlias(pattern, i + 1, path) + 1 else j = i while i < pattern.length and pattern[i] != ':' and pattern[i] != '.' ++ i n = pattern.slice(j, i) + path.wildcard |= 1 << path.length if n == '*' path.push(n) if i >= pattern.length or pattern[i] == ':' @@ -256,6 +260,8 @@ parsePath = (pattern, start, res) -> ++ i + Object.freeze(path) + if not res? return path @@ -263,6 +269,22 @@ parsePath = (pattern, start, res) -> return i +addPathMeta = (path) -> + path.alias = 0 + path.wildcard = 0 + + for p, i in path + if typeOf(p) is ARRAY_TYPE + path.alias |= 1 << i + for j in [0...p.length] by 2 + addPathMeta(p[j]) + else if p == '*' + path.wildcard |= 1 << i + + Object.freeze(path) + return path + + hammingWeight = (flags) -> flags -= ((flags >> 1) & 0x55555555) flags = (((flags >> 2) & 0x33333333) + (flags & 0x33333333)) @@ -283,4 +305,5 @@ exports.REGEXP_TYPE = REGEXP_TYPE exports.typeOf = typeOf exports.pathOverlap = pathOverlap exports.parsePath = parsePath +exports.addPathMeta = addPathMeta exports.hammingWeight = hammingWeight diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index 13aa1cf..0e15397 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -127,8 +127,9 @@ applyPresets = (currentRequest) -> return sendRpcRequest(currentRequest, id, rpcRequest) for k, v of parameters - if (vv = device.getAll(currentRequest.sessionData.deviceData, v)[0]?[8]?[0]) - parameters[k] = vv + unpacked = device.unpack(currentRequest.sessionData.deviceData, v) + if unpacked[0] and (vv = deviceData.values.value.get(unpacked[0]))? + parameters[k] = vv[0] for p in presets if query.test(parameters, p.precondition) @@ -272,8 +273,9 @@ listener = (httpRequest, httpResponse) -> return body stream.on('end', () -> - getSession(httpRequest, (sessionId, sessionData) -> - cwmpRequest = soap.request(httpRequest, session?.cwmpVersion) + cwmpRequest = null + getSession(httpRequest, f = (sessionId, sessionData) -> + cwmpRequest ?= soap.request(httpRequest, sessionData?.cwmpVersion) if not sessionData? if cwmpRequest.methodRequest?.type isnt 'Inform' httpResponse.writeHead(400) @@ -281,8 +283,11 @@ listener = (httpRequest, httpResponse) -> return deviceId = common.generateDeviceId(cwmpRequest.methodRequest.deviceId) - sessionData = session.init(deviceId, cwmpRequest.cwmpVersion, cwmpRequest.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId)) - httpRequest.connection.setTimeout(sessionData.timeout * 1000) + return session.init(deviceId, cwmpRequest.cwmpVersion, cwmpRequest.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId), (err, sessionData) -> + throw err if err + httpRequest.connection.setTimeout(sessionData.timeout * 1000) + f(null, sessionData) + ) currentRequest = { httpRequest : httpRequest, diff --git a/lib/db.coffee b/lib/db.coffee index 22b2b07..a1841a7 100644 --- a/lib/db.coffee +++ b/lib/db.coffee @@ -452,7 +452,7 @@ saveDevice = (deviceId, deviceData, isNew, callback) -> # Set object's children timestamps iter = deviceData.timestamps.exist.diff() while not (diff = iter.next()).done - continue if diff.value[0].wildcards != (1 << (diff.value[0].length - 1)) + continue if diff.value[0].wildcard != (1 << (diff.value[0].length - 1)) continue if diff.value[0][0] in ['Events', 'DeviceID', 'Tags'] if not diff.value[2]? and diff.value[1]? diff --git a/lib/device.js b/lib/device.js index 7a9508e..9e05bd4 100644 --- a/lib/device.js +++ b/lib/device.js @@ -61,12 +61,6 @@ function sanitizeParameterValue(parameterValue) { } -function getAliasDeclarations(path) { - // TODO Return declarations necessary for unpacking aliases - return []; -} - - function inferTimestamps(deviceData, path) { var res = {}; @@ -79,14 +73,14 @@ function inferTimestamps(deviceData, path) { } for (let i = path.length - 1; i > 0; -- i) { - if (path.wildcards & ((2 << i) - 1) === 0) + if (path.wildcard & ((2 << i) - 1) === 0) break; let param = null; let ancestor = path.slice(0, i); let ancestorTimestamp = 0; for (let sup of deviceData.paths.superset(ancestor)) { - if (!sup.wildcards) + if (!sup.wildcard) param = sup; let t = deviceData.timestamps.exist.get(sup); @@ -108,7 +102,7 @@ function inferTimestamps(deviceData, path) { newt[attrName] = ancestorTimestamp; for (let sub of deviceData.paths.subset(ancestor)) { - if (sub.wildcards) + if (sub.wildcard) continue; let subt = inferTimestamps(deviceData, sub.concat(path.slice(sub.length, path.length))); for (let attrName in newt) { @@ -128,9 +122,84 @@ function inferTimestamps(deviceData, path) { } +function getAliasDeclarations(path, timestamp) { + var decs = []; + + function recursive(pattern, prefix) { + var pattern = prefix.concat(pattern); + + for (let i = prefix.length; i < pattern.length; ++ i) { + if (Array.isArray(pattern[i])) { + let pat = pattern[i]; + pattern[i] = '*'; + for (let j = 0; j < pat.length; j += 2) + recursive(pat[j], pattern.slice(0, i + 1)); + } + } + + if (prefix.length) + decs.push([pattern, {exist: timestamp, value: timestamp}]); + else + decs.push([pattern, {exist: timestamp}]); + } + + recursive(path, []); + + return decs; +} + + function unpack(deviceData, path, revision) { - // TODO implement alias unpacking - return [path]; + let allMatches = []; + if (path.alias == 0) { + for (let p of deviceData.paths.subset(path, 0)) + if (deviceData.values.exist.get(p, revision)) + allMatches.push(p); + return allMatches; + } + + let wildcardPath = path.slice(); + wildcardPath.wildcard = path.wildcard; + for (let i = 0; i < wildcardPath.length; ++ i) + if (Array.isArray(wildcardPath[i])) { + wildcardPath[i] = '*'; + wildcardPath.wildcard |= 1 << i; + } + + for (let p of deviceData.paths.subset(wildcardPath, 0)) + if (deviceData.values.exist.get(p, revision)) + allMatches.push(p); + + for (let i = path.length - 1; i >= 0; -- i) { + if (Array.isArray(path[i])) { + for (let j = 0; j < path[i].length; j += 2) { + let p = wildcardPath.slice(0, i + 1).concat(path[i][j]); + p.alias = (wildcardPath.alias & ((1 << (i + 1)) - 1)) + path[i][j].alias << (i + 1); + p.wildcard = (wildcardPath.wildcard & ((1 << (i + 1)) - 1)) + path[i][j].wildcard << (i + 1); + let unpacked = unpack(deviceData, p, revision); + let filtered = []; + for (let up of unpacked) { + let v = deviceData.values.value.get(up, revision); + if (v && sanitizeParameterValue([path[i][j + 1], v[1]])[0] == v[0]) { + for (let m = 0; m < allMatches.length; ++ m) { + let k; + for (k = i; k >= 0; -- k) + if (allMatches[m][k] != up[k]) + break + + if (k < 0) { + filtered.push(allMatches[m]); + allMatches[m] = []; + } + } + } + } + allMatches = filtered; + } + } + } + + return allMatches; } @@ -176,7 +245,7 @@ function set(deviceData, path, timestamps, values) { // TODO Allow overwriting attribute value with same timestamp path = deviceData.paths.add(path); - if (!path.wildcards) { + if (!path.wildcard) { normalizeAttributes(timestamps, values); if (values.exist && path.length > 1) set(deviceData, path.slice(0, path.length - 1), @@ -187,7 +256,7 @@ function set(deviceData, path, timestamps, values) { var current = inferTimestamps(deviceData, path); for (let c in current) - if (current[c] > timestamps[c] || (current[c] == timestamps[c] && path.wildcards > 0))//!deviceData.values[c].has(path))) + if (current[c] > timestamps[c] || (current[c] == timestamps[c] && path.wildcard > 0)) delete timestamps[c]; if (timestamps.exist != null) { @@ -199,19 +268,19 @@ function set(deviceData, path, timestamps, values) { current = inferTimestamps(deviceData, path); for (let c in current) - if (current[c] > timestamps[c] || (current[c] == timestamps[c] && path.wildcards > 0))//!deviceData.values[c].has(path))) + if (current[c] > timestamps[c] || (current[c] == timestamps[c] && path.wildcard > 0)) delete timestamps[c]; let toInvalidate = new Map(); for (let sub of deviceData.paths.subset(path, 0)) { - if (path.wildcards == sub.wildcards) + if (path.wildcard == sub.wildcard) continue; for (let attrName in timestamps) { if (deviceData.timestamps[attrName].get(sub) <= timestamps[attrName]) { - if (sub.wildcards || !deviceData.values[attrName].has(sub)) + if (sub.wildcard || !deviceData.values[attrName].has(sub)) deviceData.timestamps[attrName].delete(sub); else if (deviceData.timestamps[attrName].get(sub) < timestamps[attrName]) { deviceData.timestamps[attrName].delete(sub); @@ -263,7 +332,7 @@ function set(deviceData, path, timestamps, values) { // depth = 99; // // for (let sup of deviceData.paths.superset(path, depth)) { - // if ((sup.wildcards && (3 << path.length) - 1) != path.wildcards) { + // if ((sup.wildcard && (3 << path.length) - 1) != path.wildcard) { // throw new Error('Not implemented'); // } // } diff --git a/lib/path-set.js b/lib/path-set.js index e7ab8b7..2cf117c 100644 --- a/lib/path-set.js +++ b/lib/path-set.js @@ -79,18 +79,19 @@ class PathSet { } } - var wildcards = 0; + var wildcard = 0; this.lengthIndex[path.length].add(path); for (let i = 0; i < path.length; ++ i) { let fragment = path[i]; if (fragment === '*') - wildcards |= 1 << i; + wildcard |= 1 << i; this.fragmentIndex[i][fragment].add(path); } - if (path.wildcards == null) { - path.wildcards = wildcards; + if (path.wildcard == null) { + path.alias = 0; + path.wildcard = wildcard; Object.freeze(path); } return path; diff --git a/lib/sandbox.coffee b/lib/sandbox.coffee index 14b0fbd..3621a52 100644 --- a/lib/sandbox.coffee +++ b/lib/sandbox.coffee @@ -83,23 +83,24 @@ sandbox.context.declare = (decs) -> res = {} for k, v of decs path = common.parsePath(k) + single = true + for p in path + if p == '*' or common.typeOf(p) isnt common.STRING_TYPE + single = false + break - res[k] = {} unpacked = device.unpack(sandbox.deviceData, path, sandbox.revision) - for path in unpacked - iter = sandbox.deviceData.paths.subset(path) - while (param = iter.next().value)? - if param.wildcards == 0 and sandbox.deviceData.values.exist.get(param, sandbox.revision) - r = {} - for attrName, dec of v - if (attrValue = sandbox.deviceData.values[attrName].get(param, sandbox.revision))? - r[attrName] = [sandbox.deviceData.timestamps[attrName].get(param, sandbox.revision), attrValue] + for param in unpacked + r = {} + for attrName, dec of v + if (attrValue = sandbox.deviceData.values[attrName].get(param, sandbox.revision))? + r[attrName] = [sandbox.deviceData.timestamps[attrName].get(param, sandbox.revision), attrValue] - p = param.join('.') - if p == k - res[k] = r - else - res[k][p] = r + if single + res[k] = r + else + res[k] ?= {} + res[k][param.join('.')] = r return res diff --git a/lib/session.coffee b/lib/session.coffee index b751876..d44dc7a 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -33,15 +33,11 @@ initDeviceData = () -> paths : new PathSet() timestamps : {exist: new VersionedMap(), object: new VersionedMap(), writable: new VersionedMap(), value: new VersionedMap()} values : {exist: new VersionedMap(), object: new VersionedMap(), writable: new VersionedMap(), value: new VersionedMap()} - declarationTimestamps : new Map() - declarationValues : new Map() - virtualParameterTimestamps : new Map() - virtualParameterValues : new Map() loaded : new Map() } -init = (deviceId, cwmpVersion, timeout) -> +init = (deviceId, cwmpVersion, timeout, callback) -> timestamp = Date.now() sessionData = { @@ -55,22 +51,33 @@ init = (deviceId, cwmpVersion, timeout) -> rpcCount: 0 cycle: 0 extensionsCache: [] + declarations: [] } - return sessionData + cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> + return callback(err) if err + + sessionData.cache = { + hash: hash + provisions: provisions + virtualParameters: virtualParameters + } + + return callback(null, sessionData) + ) -loadParameters = (sessionData, toLoad, virtualParameters, callback) -> - if not toLoad.length +loadParameters = (sessionData, callback) -> + if not sessionData.toLoad?.length return callback() - db.fetchDevice(sessionData.deviceId, sessionData.timestamp, toLoad, (err, parameters, loaded) -> + db.fetchDevice(sessionData.deviceId, sessionData.timestamp, sessionData.toLoad, (err, parameters, loaded) -> return callback(err) if err if not parameters? # Device not available in database, mark as new sessionData.new = true - path = sessionData.deviceData.paths.add([]) + path = sessionData.deviceData.paths.add(['*']) sessionData.deviceData.loaded.set(path, 99) return callback() @@ -91,7 +98,7 @@ loadParameters = (sessionData, toLoad, virtualParameters, callback) -> sessionData.deviceData.values[k].set(path, v, 0) # Virtual parameters - for p in toLoad + for p in sessionData.toLoad p = sessionData.deviceData.paths.add(p) if p[0] == '*' or p[0] == 'VirtualParameters' if p.length == 1 @@ -105,18 +112,19 @@ loadParameters = (sessionData, toLoad, virtualParameters, callback) -> if not p[1]? sessionData.deviceData.timestamps.exist.set(p, sessionData.timestamp, 0) - for k, v of virtualParameters + for k, v of sessionData.cache.virtualParameters p = sessionData.deviceData.paths.add(p) sessionData.deviceData.timestamps.exist.set(p, sessionData.timestamp, 0) sessionData.deviceData.timestamps.object.set(p, sessionData.timestamp, 0) sessionData.deviceData.values.exist.set(p, 1, 0) sessionData.deviceData.values.object.set(p, 0, 0) - else if virtualParameters[p[1]]? + else if sessionData.cache.virtualParameters[p[1]]? sessionData.deviceData.timestamps.exist.set(p, sessionData.timestamp, 0) sessionData.deviceData.timestamps.object.set(p, sessionData.timestamp, 0) sessionData.deviceData.values.exist.set(p, 1, 0) sessionData.deviceData.values.object.set(p, 0, 0) + sessionData.toLoad = null return callback() ) @@ -163,26 +171,10 @@ inform = (sessionData, rpcReq, callback) -> {exist: timestamp, object: timestamp, writable: timestamp, value: timestamp}, {exist: 1, object: 0, writable: 0, value: [timestamp, 'xsd:dateTime']}]) - toLoad = [] for p in params - tl = [] - for i in [p[0].length...0] by -1 - path = p[0].slice(0, i) - loaded = 0 - iter = sessionData.deviceData.paths.superset(path) - while sup = iter.next().value - if (l = sessionData.deviceData.loaded.get(sup)) > loaded - loaded = l + loadPath(sessionData, p[0]) - if loaded - tl = tl.slice(0, tl.length - (loaded - 1)) - break - - tl.push(path) - - toLoad = toLoad.concat(tl) - - loadParameters(sessionData, toLoad, null, (err) -> + loadParameters(sessionData, (err) -> return callback(err) if err v.revision = 1 for k, v of sessionData.deviceData.timestamps @@ -217,213 +209,250 @@ clearProvisions = (sessionData) -> sessionData.extensionsCache.length = 0 +runProvisions = (sessionData, callback) -> + done = true + _extensions = [] + allDeclarations = [] + for provision in sessionData.provisions + if not sessionData.cache.provisions[provision[0]]? + switch provision[0] + when 'refresh' + path = common.parsePath(provision[1]).slice() + l = path.length + path.length = 10 + path.fill('*', l) + t = provision[2] + t += sessionData.timestamp if t <= 0 -generateRpcRequest = (sessionData) -> - if sessionData.deviceData.declarationTimestamps.size == 0 and - sessionData.deviceData.declarationValues.size == 0 - return null + for i in [l...path.length] by 1 + p = common.addPathMeta(path.slice(0, i)) + allDeclarations.push([p, {exist: t, object: 1, writable: 1, value: t}]) + when 'value' + allDeclarations.push([common.parsePath(provision[1]), {exist: 1, value: 1}, {value: [provision[2]]}]) + when 'tag' + allDeclarations.push([['Tags', provision[1]], {exist: 1, value: 1}, {exist: 1, value: [provision[2], 'xsd:boolean']}]) + when '_task' + # A special provision for tasks compatibility + switch provision[2] + when 'getParameterValues' + for i in [3...provision.length] by 1 + allDeclarations.push([common.parsePath(provision[i]), {exist: 1, value: sessionData.timestamp}]) + when 'setParameterValues' + for i in [3...provision.length] by 3 + v = if provision[i + 2] then [provision[i + 1], provision[i + 2]] else [provision[i + 1]] + allDeclarations.push([common.parsePath(provision[i]), {exist: 1, value: 1}, {value: v}]) + when 'refreshObject' + path = common.parsePath(provision[3]).slice() + l = path.length + path.length = 10 + path.fill('*', l) + for i in [l...path.length] by 1 + p = common.addPathMeta(path.slice(0, i)) + allDeclarations.push([p, {exist: sessionData.timestamp, object: 1, writable: 1, value: sessionData.timestamp}]) + continue - refreshPaths = new PathSet() - refreshAttributes = { - exist: new Set() - object: new Set() - writable: new Set() - value: new Set() - } - setValues = new Map() - refreshGpn = new Map() + ret = sandbox.run(sessionData.cache.provisions[provision[0]].script, provision.slice(1), sessionData.timestamp, sessionData.deviceData, sessionData.extensionsCache, 0, sessionData.revisions[0] >> 1) + if ret.extensions?.length + _extensions = _extensions.concat(ret.extensions) + done &&= ret.done + allDeclarations = allDeclarations.concat(ret.declarations) + + if _extensions.length + return runExtensions(sessionData, (sessionData.revisions[0] >> 1), _extensions, (err) -> + return callback(err) if err + return runProvisions(sessionData, callback) + ) + + for d in allDeclarations + state = applyDeclaration(sessionData, d[0], 0, d[1], d[2]) + sessionData.declarations.push([d[0], state, d[1], d[2]]) + + return callback(null, done) + + +runVirtualParameters = (sessionData, inception, callback) -> + if not sessionData.syncState? or Object.keys(sessionData.syncState.virtualParameters).length == 0 + return callback() + + + virtualParameterDeclarations = sessionData.syncState.virtualParameters + sessionData.syncState.virtualParameters = {} revision = sessionData.revisions.reduce(((a, b) -> a + (b >> 1)), 0) + 1 + sessionData.revisions[inception] ?= 0 + decs = [] + virtualParameterUpdates = {} + _extensions = [] - test = (deviceData, path, timestamps, values) -> - iter = deviceData.paths.superset(path) - param = null - existTimestamp = 0 - while sup = iter.next().value - param = sup if not sup.wildcards - for k, v of timestamps - continue if not (t = deviceData.timestamps[k].get(sup, revision))? - existTimestamp = Math.max(existTimestamp, t) if k is 'exist' - delete timestamps[k] if t >= v + firstRevision = revision - (sessionData.revisions[sessionData.revisions.length - 1] >> 1) + for vpName, vpDeclarations of virtualParameterDeclarations + ret = sandbox.run(sessionData.cache.virtualParameters[vpName].script, [vpDeclarations.timestamps, vpDeclarations.values], sessionData.timestamp, sessionData.deviceData, sessionData.extensionsCache, firstRevision, revision) + if ret.extensions?.length + _extensions = _extensions.concat(ret.extensions) - if not Object.keys(timestamps).length and - not (values? and Object.keys(values).length) - return + decs = decs.concat(ret.declarations) - if param? - if deviceData.values.exist.get(param, revision) - param = refreshPaths.add(param) - isObject = deviceData.values.object.get(param, revision) - isWritable = deviceData.values.writable.get(param, revision) - for k of timestamps - if k isnt 'value' or isObject == 0 - refreshAttributes[k].add(param) - else if not isObject? - refreshAttributes.object.add(param) + if ret.done + virtualParameterUpdates[vpName] = ret.returnValue - if values?.value? - if isObject == 0 and isWritable == 1 - curVal = deviceData.values.value.get(param, revision) - cur = setValues.get(param) - if not cur? - cur = [deviceData.values.value.get(param, revision), values.value] - setValues.set(param, cur) - else - cur[1] = values.value - else - if not isObject? - refreshAttributes.object.add(param) - if not isWritable? - refreshAttributes.writable.add(param) - return - else if existTimestamp >= (timestamps.exist or 1) - return - else if path.indexOf('*') >= 0 - subIter = deviceData.paths.subset(path) - while sub = subIter.next().value - continue if not deviceData.values.exist.get(sub, revision) + if _extensions.length + return runExtensions(sessionData, revision, _extensions, (err) -> + return callback(err) if err + return runVirtualParameters(sessionData, inception, callback) + ) - isObject = deviceData.values.object.get(sub, revision) - isWritable = deviceData.values.writable.get(sub, revision) - - for k, v of timestamps - if not (v <= deviceData.timestamps[k].get(sub, revision)) - if k isnt 'value' or isObject == 0 - sub = refreshPaths.add(sub) - refreshAttributes[k].add(sub) - else if not isObject? - sub = refreshPaths.add(sub) - refreshAttributes.object.add(sub) - - if values?.value? - if isObject == 0 and isWritable == 1 - curVal = deviceData.values.value.get(sub, revision) - cur = setValues.get(sub) - if not cur? - cur = [deviceData.values.value.get(sub, revision), values.value] - setValues.set(sub, cur) - else - cur[1] = values.value - else - if not isObject? - refreshAttributes.object.add(sub) - if not isWritable? - refreshAttributes.writable.add(sub) - - if existTimestamp >= (timestamps.exist or 1) - return - - childrenTimestamp = existTimestamp - - for i in [path.length - 1...0] by -1 - existTimestamp = 0 - p = path.slice(0, i) - iter = deviceData.paths.superset(p) - param = null - while sup = iter.next().value - param = sup if not sup.wildcards - if (t = deviceData.timestamps.exist.get(sup, revision))? - existTimestamp = Math.max(existTimestamp, t) - - if param? - if deviceData.values.exist.get(param, revision) - isObject = deviceData.values.object.get(param, revision) - objectTimestamp = deviceData.timestamps.object.get(param, revision) - if isObject - if childrenTimestamp == 0 - param = refreshPaths.add(param) - refreshGpn.set(param, refreshGpn.get(param) | ((1 << path.length - i) - 1)) - else if childrenTimestamp < (timestamps.exist or 1) - param = refreshPaths.add(param) - refreshGpn.set(param, refreshGpn.get(param) or 1) - if not (objectTimestamp >= (timestamps.exist or 1)) - param = refreshPaths.add(param) - refreshAttributes.object.add(param) - return - else if existTimestamp >= (timestamps.exist or 1) - return - else if p.indexOf('*') >= 0 - subIter = deviceData.paths.subset(p) - ret = true - while sub = subIter.next().value - continue if not deviceData.values.exist.get(sub, revision) - # TODO check if not object - pp = sub.concat(path.slice(sub.length)) - ts = {} - ts[k] = v for k, v of timestamps - ret = ret and test(deviceData, pp, ts) - - if existTimestamp >= (timestamps.exist or 1) - return - - childrenTimestamp = existTimestamp - - p = refreshPaths.add([]) - refreshGpn.set(p, refreshGpn.get(p) or 1) - return false - - iter = sessionData.deviceData.declarationTimestamps.entries() - while (pair = iter.next().value) - timestamps = {} - timestamps[k] = v for k, v of pair[1] - test(sessionData.deviceData, pair[0], timestamps, sessionData.deviceData.declarationValues.get(pair[0])) + for d in decs + state = applyDeclaration(sessionData, d[0], 0, d[1], d[2]) + sessionData.declarations.push([d[0], state, d[1], d[2]]) - iter = refreshAttributes.exist.values() + if Object.keys(virtualParameterUpdates).length == Object.keys(virtualParameterDeclarations).length + + rev = sessionData.revisions.reduce(((a, b) -> a + (b >> 1)), 0) + 1 + v.revision = rev for k, v of sessionData.deviceData.timestamps + v.revision = rev for k, v of sessionData.deviceData.values + for vpName, vpDeclarations of virtualParameterDeclarations + commitVirtualParameter(sessionData, vpName, vpDeclarations, virtualParameterUpdates[vpName]) + + if sessionData.revisions[inception] > 0 + sessionData.revisions.length = inception + rev = sessionData.revisions.reduce(((a, b) -> a + (b >> 1)), 0) + 1 + m.collapse(rev) for k, m of sessionData.deviceData.timestamps + m.collapse(rev) for k, m of sessionData.deviceData.values + if sessionData.extensionsCache.length > rev + sessionData.extensionsCache.length = rev + + return runVirtualParameters(sessionData, inception + 1, callback) + + +rpcRequest = (sessionData, declarations, callback) -> + revision = sessionData.revisions.reduce(((a, b) -> a + (b >> 1)), 0) + 1 + + v.revision = revision for k, v of sessionData.deviceData.timestamps + v.revision = revision for k, v of sessionData.deviceData.values + + f = (cb) -> + return cb() if not sessionData.declarations.length + + for d, i in sessionData.declarations + state = applyDeclaration(sessionData, d[0], d[1], d[2], d[3]) + d[1] = state + + if sessionData.toLoad?.length + return loadParameters(sessionData, (err) -> + return cb(err) if err + return f(cb) + ) + + runVirtualParameters(sessionData, 1, (err) -> + return cb(err) if err + + if sessionData.toLoad?.length + return loadParameters(sessionData, (err) -> + return cb(err) if err + return f(cb) + ) + + return cb(null, generateRpcRequest(sessionData)) + ) + + if declarations? + for d in declarations + sessionData.declarations.push([d[0], 0, d[1], d[2]]) + + if sessionData.declarations.length + return f((err, rpcReq) -> + return callback(err) if err + if rpcReq? + sessionData.declarations = [] if rpcReq.done + sessionData.rpcRequest = rpcReq + return callback(null, generateRpcId(sessionData), rpcReq) + + sessionData.declarations = [] + ++ sessionData.revisions[sessionData.revisions.length - 1] + return rpcRequest(sessionData, null, callback) + ) + + if not sessionData.provisions.length + return callback() + + runProvisions(sessionData, (err, done) -> + return callback(err) if err + return f((err, rpcReq) -> + return callback(err) if err + + if rpcReq? + sessionData.declarations = [] if rpcReq.done + sessionData.rpcRequest = rpcReq + return callback(null, generateRpcId(sessionData), rpcReq) + + sessionData.declarations = [] + if done + return callback() + ++ sessionData.revisions[sessionData.revisions.length - 1] + return rpcRequest(sessionData, null, callback) + ) + ) + + +generateRpcRequest = (sessionData) -> + syncState = sessionData.syncState + + iter = syncState.refreshAttributes.exist.values() while (path = iter.next().value) - descendantIter = refreshPaths.subset(path, 99) + descendantIter = syncState.paths.subset(path, 99) found = false while (p = descendantIter.next().value)? - if refreshAttributes.value.has(p) or - refreshAttributes.object.has(p) or - refreshAttributes.writable.has(p) or - refreshGpn.has(p) + if syncState.refreshAttributes.value.has(p) or + syncState.refreshAttributes.object.has(p) or + syncState.refreshAttributes.writable.has(p) or + syncState.refreshGpn.has(p) found = true break if not found - path = refreshPaths.add(path.slice(0, -1)) - refreshGpn.set(path, refreshGpn.get(path) or 1) + path = syncState.paths.add(path.slice(0, -1)) + syncState.refreshGpn.set(path, syncState.refreshGpn.get(path) or 1) - iter = refreshAttributes.object.values() + iter = syncState.refreshAttributes.object.values() while (path = iter.next().value) - descendantIter = refreshPaths.subset(path, 99) + descendantIter = syncState.paths.subset(path, 99) found = false while (p = descendantIter.next().value)? - if refreshAttributes.value.has(p) or + if syncState.refreshAttributes.value.has(p) or (p.length > path.length and - (refreshAttributes.object.has(p) or - refreshAttributes.writable.has(p))) + (syncState.refreshAttributes.object.has(p) or + syncState.refreshAttributes.writable.has(p))) found = true break if not found - path = refreshPaths.add(path.slice(0, -1)) - refreshGpn.set(path, refreshGpn.get(path) or 1) + path = syncState.paths.add(path.slice(0, -1)) + syncState.refreshGpn.set(path, syncState.refreshGpn.get(path) or 1) - iter = refreshAttributes.writable.values() + iter = syncState.refreshAttributes.writable.values() while (path = iter.next().value) - path = refreshPaths.add(path.slice(0, -1)) - refreshGpn.set(path, refreshGpn.get(path) or 1) + path = syncState.paths.add(path.slice(0, -1)) + syncState.refreshGpn.set(path, syncState.refreshGpn.get(path) or 1) - iter = refreshGpn.keys() + iter = syncState.refreshGpn.keys() while (path = iter.next().value) - descendantIter = refreshPaths.subset(path, 99) + descendantIter = syncState.paths.subset(path, 99) found = false while (p = descendantIter.next().value)? continue if p == path - if (flags = refreshGpn.get(p)) + if (flags = syncState.refreshGpn.get(p)) flags <<= p.length - path.length - refreshGpn.set(path, refreshGpn.get(path) | flags) - refreshGpn.delete(p) + syncState.refreshGpn.set(path, syncState.refreshGpn.get(path) | flags) + syncState.refreshGpn.delete(p) rpcReq = null completed = true - if refreshGpn.size + if syncState.refreshGpn.size GET_PARAMETER_NAMES_DEPTH_THRESHOLD = config.get('GET_PARAMETER_NAMES_DEPTH_THRESHOLD', sessionData.deviceId) - pair = refreshGpn.entries().next().value + pair = syncState.refreshGpn.entries().next().value nextLevel = true if pair[0].length >= GET_PARAMETER_NAMES_DEPTH_THRESHOLD @@ -431,7 +460,7 @@ generateRpcRequest = (sessionData) -> else if common.hammingWeight(pair[1] >> 3) >= 3 nextLevel = false - if refreshGpn.size > 1 or (pair[1] > 1 and nextLevel) + if syncState.refreshGpn.size > 1 or (pair[1] > 1 and nextLevel) completed = false rpcReq = { @@ -440,7 +469,7 @@ generateRpcRequest = (sessionData) -> nextLevel: nextLevel } - if refreshAttributes.value.size + if syncState.refreshAttributes.value.size if rpcReq? completed = false else @@ -448,12 +477,12 @@ generateRpcRequest = (sessionData) -> config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId) parameterNames = [] - iter = refreshAttributes.value.values() + iter = syncState.refreshAttributes.value.values() while (path = iter.next().value) and parameterNames.length < TASK_PARAMETERS_BATCH_SIZE parameterNames.push(path) - if refreshAttributes.value.size > parameterNames.length + if syncState.refreshAttributes.value.size > parameterNames.length completed = false rpcReq = { @@ -465,7 +494,7 @@ generateRpcRequest = (sessionData) -> config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId) parameterValues = [] - setValues.forEach((v, k) -> + syncState.setValues.forEach((v, k) -> return if completed == false if rpcReq? @@ -489,87 +518,273 @@ generateRpcRequest = (sessionData) -> parameterList: ([p[0].join('.'), p[1], p[2]] for p in parameterValues) } - if completed - sessionData.deviceData.declarationTimestamps.clear() - sessionData.deviceData.declarationValues.clear() + if completed and rpcReq? + rpcReq.done = true return rpcReq -getParametersToLoad = (sessionData) -> - toLoad = [] +processVirtualparameterDeclaration = (sessionData, path, timestamps, values) -> + ret = true + return ret if path.length != 2 + deviceData = sessionData.deviceData + subIter = sessionData.deviceData.paths.subset(path) + while sub = subIter.next().value + continue if not deviceData.values.exist.get(sub) - test = (timestamps, path) -> - tl = [] - for i in [path.length...0] by -1 - p = path.slice(0, i) - loaded = 0 - iter = sessionData.deviceData.paths.superset(p, 1) + isWritable = deviceData.values.writable.get(sub) - while sup = iter.next().value - if (l = sessionData.deviceData.loaded.get(sup)) > loaded - loaded = l + for k, v of timestamps + if not (v <= deviceData.timestamps[k].get(sub)) + sessionData.syncState.virtualParameters[sub[1]] ?= {timestamps: {}, values: {}} + sessionData.syncState.virtualParameters[sub[1]].timestamps[k] = + Math.max(v, sessionData.syncState.virtualParameters[sub[1]].timestamps[k] ? 0) + ret = false - if loaded - tl = tl.slice(0, tl.length - (loaded - 1)) - break - - tl.push(p) - - toLoad = toLoad.concat(tl) - - sessionData.deviceData.declarationTimestamps.forEach(test) - sessionData.deviceData.virtualParameterTimestamps.forEach(test) - - return toLoad - - -extractVirtualParameterDeclarations = (deviceData, revision, virtualParameters) -> - if deviceData.virtualParameterTimestamps.size == 0 and - deviceData.virtualParameterValues.size == 0 - return {} - - iter = deviceData.paths.subset(['VirtualParameters', '*']) - virtualParameterDeclarations = {} - while path = iter.next().value - continue if path.wildcards or not virtualParameters[path[1]]? - - if declarationTimestamps = deviceData.virtualParameterTimestamps.get(path) - for k, v of declarationTimestamps - ct = deviceData.timestamps[k].get(path, revision) - if not (v <= ct) - virtualParameterDeclarations[path[1]] ?= {timestamps: {}, values: {}} - virtualParameterDeclarations[path[1]].timestamps[k] = - Math.max(v, virtualParameterDeclarations[path[1]].timestamps[k] ? 0) - - if declarationValues = deviceData.virtualParameterValues.get(path) - for k, v of declarationValues - virtualParameterDeclarations[path[1]] ?= {timestamps: {}, values: {}} - cv = deviceData.values[k].get(path, revision) - if not virtualParameterDeclarations[path[1]].values[k]? - virtualParameterDeclarations[path[1]].values[k] = [cv, v] + if values?.value? + if isWritable == 1 + sessionData.syncState.virtualParameters[sub[1]] ?= {timestamps: {}, values: {}} + cur = sessionData.syncState.virtualParameters[sub[1]].values.value + if not cur? + cur = [deviceData.values.value.get(sub), values.value] + sessionData.syncState.virtualParameters[sub[1]].values.value = cur else - virtualParameterDeclarations[path[1]].values[k][1] = v + cur[1] = values.value + else if not isWritable? + sessionData.syncState.virtualParameters[sub[1]] ?= {timestamps: {}, values: {}} + sessionData.syncState.virtualParameters[sub[1]].timestamps.writable = + Math.max(1, sessionData.syncState.virtualParameters[sub[1]].timestamps.writable ? 0) + ret = false - for k, v of virtualParameterDeclarations - for attrName, attrValue of v.values - val = attrValue[1].slice() - val[1] = attrValue[0][1] if not val[1]? - device.sanitizeParameterValue(val) - if val[0] != attrValue[0][0] or val[1] != attrValue[0][1] - v.values[attrName] = attrValue[1] - else - delete v.values[attrName] - if Object.keys(v.values).length == 1 - if Object.keys(v.timestamps).length == 0 - delete virtualParameterDeclarations[k] + return ret + + +processDeclaration = (sessionData, path, _timestamps, values) -> + sessionData.syncState ?= { + paths: new PathSet() + refreshAttributes: { + exist: new Set() + object: new Set() + writable: new Set() + value: new Set() + } + setValues: new Map() + refreshGpn: new Map() + virtualParameters: {} + } + + syncState = sessionData.syncState + deviceData = sessionData.deviceData + + timestamps = {} + timestamps[k] = v for k, v of _timestamps + + ret = true + + if path[0] == '*' or path[0] == 'VirtualParameters' + ret &&= processVirtualparameterDeclaration(sessionData, path, _timestamps, values) + return ret if path[0] != '*' + + iter = sessionData.deviceData.paths.superset(path) + param = null + existTimestamp = 0 + while sup = iter.next().value + param = sup if not sup.wildcard + for k, v of timestamps + continue if not (t = deviceData.timestamps[k].get(sup))? + existTimestamp = Math.max(existTimestamp, t) if k is 'exist' + delete timestamps[k] if t >= v + + if not Object.keys(timestamps).length and + not (values? and Object.keys(values).length) + return ret + + if param? + if deviceData.values.exist.get(param) + param = syncState.paths.add(param) + isObject = deviceData.values.object.get(param) + isWritable = deviceData.values.writable.get(param) + + for k of timestamps + if k isnt 'value' or isObject == 0 + syncState.refreshAttributes[k].add(param) + ret = false + else if not isObject? + syncState.refreshAttributes.object.add(param) + ret = false + + if values?.value? + if isObject == 0 and isWritable == 1 + cur = syncState.setValues.get(param) + if not cur? + cur = [deviceData.values.value.get(param), values.value] + syncState.setValues.set(param, cur) else - delete v.values[attrName] + cur[1] = values.value + else + if not isObject? + syncState.refreshAttributes.object.add(param) + ret = false + if not isWritable? + syncState.refreshAttributes.writable.add(param) + ret = false + return ret + else if existTimestamp >= (timestamps.exist or 1) + return ret + else if path.indexOf('*') >= 0 + subIter = deviceData.paths.subset(path) + while sub = subIter.next().value + continue if not deviceData.values.exist.get(sub) - deviceData.virtualParameterTimestamps.clear() - deviceData.virtualParameterValues.clear() + isObject = deviceData.values.object.get(sub) + isWritable = deviceData.values.writable.get(sub) - return virtualParameterDeclarations + for k, v of timestamps + if not (v <= deviceData.timestamps[k].get(sub)) + if k isnt 'value' or isObject == 0 + sub = syncState.paths.add(sub) + syncState.refreshAttributes[k].add(sub) + ret = false + else if not isObject? + sub = syncState.paths.add(sub) + syncState.refreshAttributes.object.add(sub) + ret = false + + if values?.value? + if isObject == 0 and isWritable == 1 + cur = setValues.setValues.get(sub) + if not cur? + cur = [deviceData.values.value.get(sub), values.value] + setValues.setValues.set(sub, cur) + else + cur[1] = values.value + else + if not isObject? + syncState.refreshAttributes.object.add(sub) + ret = false + if not isWritable? + syncState.refreshAttributes.writable.add(sub) + ret = false + + if existTimestamp >= (timestamps.exist or 1) + return + + childrenTimestamp = existTimestamp + + for i in [path.length - 1...0] by -1 + existTimestamp = 0 + p = path.slice(0, i) + iter = deviceData.paths.superset(p) + param = null + while sup = iter.next().value + param = sup if not sup.wildcard + if (t = deviceData.timestamps.exist.get(sup))? + existTimestamp = Math.max(existTimestamp, t) + + if param? + if deviceData.values.exist.get(param) + isObject = deviceData.values.object.get(param) + objectTimestamp = deviceData.timestamps.object.get(param) + if isObject + if childrenTimestamp == 0 + param = syncState.paths.add(param) + syncState.refreshGpn.set(param, syncState.refreshGpn.get(param) | ((1 << path.length - i) - 1)) + ret = false + else if childrenTimestamp < (timestamps.exist or 1) + param = syncState.paths.add(param) + syncState.refreshGpn.set(param, syncState.refreshGpn.get(param) or 1) + ret = false + if not (objectTimestamp >= (timestamps.exist or 1)) + param = syncState.paths.add(param) + syncState.refreshAttributes.object.add(param) + ret = false + # return ret + # else if existTimestamp >= (timestamps.exist or 1) + # return ret + else if p.indexOf('*') >= 0 + subIter = deviceData.paths.subset(p) + while sub = subIter.next().value + continue if not deviceData.values.exist.get(sub) + # TODO check if not object + pp = sub.concat(path.slice(sub.length)) + ts = {} + ts[k] = v for k, v of timestamps + ret &&= processDeclaration(sessionData, pp, ts) + + if existTimestamp >= (timestamps.exist or 1) + return ret + + childrenTimestamp = existTimestamp + + p = syncState.paths.add([]) + ret = false + syncState.refreshGpn.set(p, syncState.refreshGpn.get(p) or 1) + return ret + + +loadPath = (sessionData, path) -> + tl = [] + for i in [path.length...0] by -1 + p = path.slice(0, i) + loaded = 0 + iter = sessionData.deviceData.paths.superset(p) + + while sup = iter.next().value + if (l = sessionData.deviceData.loaded.get(sup)) > loaded + loaded = l + + if loaded + tl = tl.slice(0, tl.length - (loaded - 1)) + break + + tl.push(p) + + if not tl.length + return true + + sessionData.toLoad ?= [] + sessionData.toLoad = sessionData.toLoad.concat(tl) + return false + + +applyDeclaration = (sessionData, path, state, timestamps, values) -> + if state == 100 + if not values? or Object.keys(values).length == 0 + return state + timestamps = {} + + decs = device.getAliasDeclarations(path, timestamps.exist ? 1) + + if state <= 1 + loaded = true + for d in decs + loaded &&= loadPath(sessionData, d[0]) + return 1 if not loaded + + if decs.length == 1 + satisfied = processDeclaration(sessionData, decs[0][0], timestamps, values) + if satisfied + return 100 + else + return 2 + else + if state <= 2 + satisfied = true + + for d in decs + satisfied &&= processDeclaration(sessionData, d[0], d[1], d[2]) + + return 2 if not satisfied + + unpacked = device.unpack(sessionData.deviceData, path) + satisfied = true + for up in unpacked + satisfied &&= processDeclaration(sessionData, path, timestamps, values) + + if satisfied + return 100 + else + return 4 commitVirtualParameter = (sessionData, name, declaration, update) -> @@ -626,202 +841,6 @@ runExtensions = (sessionData, revision, _extensions, callback) -> return callback() -rpcRequest = (sessionData, declarations, callback) -> - if not declarations?.length - if (rpcReq = generateRpcRequest(sessionData))? - sessionData.rpcRequest = rpcReq - return callback(null, generateRpcId(sessionData), rpcReq) - - allDeclarations = declarations?.slice() ? [] - - cache.getProvisionsAndVirtualParameters((err, presetsHash, provisions, virtualParameters) -> - return callback(err) if err or (sessionData.presetsHash? and sessionData.presetsHashpresetsHash) - - done = true - _extensions = [] - for provision in sessionData.provisions - if not provisions[provision[0]]? - switch provision[0] - when 'refresh' - path = common.parsePath(provision[1]) - l = path.length - path.length = 16 - path.fill('*', l) - t = provision[2] - t += sessionData.timestamp if t <= 0 - - for i in [l...path.length] by 1 - allDeclarations.push([path.slice(0, i), {exist: t, object: 1, writable: 1, value: t}]) - when 'value' - allDeclarations.push([common.parsePath(provision[1]), {exist: 1, value: 1}, {value: [provision[2]]}]) - when 'tag' - allDeclarations.push([['Tags', provision[1]], {exist: 1, value: 1}, {exist: 1, value: [provision[2], 'xsd:boolean']}]) - when '_task' - # A special provision for tasks compatibility - switch provision[2] - when 'getParameterValues' - for i in [3...provision.length] by 1 - allDeclarations.push([common.parsePath(provision[i]), {exist: 1, value: sessionData.timestamp}]) - when 'setParameterValues' - for i in [3...provision.length] by 3 - v = if provision[i + 2] then [provision[i + 1], provision[i + 2]] else [provision[i + 1]] - allDeclarations.push([common.parsePath(provision[i]), {exist: 1, value: 1}, {value: v}]) - when 'refreshObject' - path = common.parsePath(provision[3]) - l = path.length - path.length = 16 - path.fill('*', l) - for i in [l...16] by 1 - allDeclarations.push([path.slice(0, i), {exist: sessionData.timestamp, object: 1, writable: 1, value: sessionData.timestamp}]) - continue - - ret = sandbox.run(provisions[provision[0]].script, provision.slice(1), sessionData.timestamp, sessionData.deviceData, sessionData.extensionsCache, 0, sessionData.revisions[0] >> 1) - if ret.extensions?.length - _extensions = _extensions.concat(ret.extensions) - done &&= ret.done - allDeclarations = allDeclarations.concat(ret.declarations) - - if done and (not allDeclarations?.length or sessionData.revisions[0] > 1) - return callback() - - if _extensions.length - runExtensions(sessionData, (sessionData.revisions[0] >> 1), _extensions, (err) -> - return callback(err) if err - return rpcRequest(sessionData, declarations, callback) - ) - return - - doVirtualParameters = (inception, virtualParameterDeclarations, cb) -> - revision = sessionData.revisions.reduce(((a, b) -> a + (b >> 1)), 0) + 1 - if not virtualParameterDeclarations? - virtualParameterDeclarations = - extractVirtualParameterDeclarations(sessionData.deviceData, revision, virtualParameters) - if not Object.keys(virtualParameterDeclarations).length - return cb(false) - - sessionData.revisions[inception] ?= 0 - decs = [] - virtualParameterUpdates = {} - _extensions = [] - - firstRevision = revision - (sessionData.revisions[sessionData.revisions.length - 1] >> 1) - for vpName, vpDeclarations of virtualParameterDeclarations - ret = sandbox.run(virtualParameters[vpName].script, [vpDeclarations.timestamps, vpDeclarations.values], sessionData.timestamp, sessionData.deviceData, sessionData.extensionsCache, firstRevision, revision) - - if ret.extensions?.length - _extensions = _extensions.concat(ret.extensions) - - decs = decs.concat(ret.declarations) - if ret.done - virtualParameterUpdates[vpName] = ret.returnValue - - if _extensions.length - runExtensions(sessionData, revision, _extensions, (err) -> - return callback(err) if err - return doVirtualParameters(inception, virtualParameterDeclarations, cb) - ) - return - - if Object.keys(virtualParameterUpdates).length == Object.keys(virtualParameterDeclarations).length - rev = sessionData.revisions.reduce(((a, b) -> a + (b >> 1)), 0) + 1 - v.revision = rev for k, v of sessionData.deviceData.timestamps - v.revision = rev for k, v of sessionData.deviceData.values - for vpName, vpDeclarations of virtualParameterDeclarations - commitVirtualParameter(sessionData, vpName, vpDeclarations, virtualParameterUpdates[vpName]) - - if sessionData.revisions[inception] == 0 - return cb(false) - - sessionData.revisions.length = inception - rev = sessionData.revisions.reduce(((a, b) -> a + (b >> 1)), 0) + 1 - m.collapse(rev) for k, m of sessionData.deviceData.timestamps - m.collapse(rev) for k, m of sessionData.deviceData.values - if sessionData.extensionsCache.length > rev - sessionData.extensionsCache.length = rev - return cb(true) - - applyDeclarations(inception, decs, () -> - return doVirtualParameters(inception, virtualParameterDeclarations, cb) - ) - - applyDeclarations = (inception, decs, cb) -> - declare = (path, timestamps, values, allTimestamps, allValues) -> - dt = null - for k, v of timestamps - if not dt? - dt = allTimestamps.get(path) - if not dt? - dt = {} - allTimestamps.set(path, dt) - v += sessionData.timestamp if v <= 0 - dt[k] = v if v > (dt[k] || 0) - - dv = null - for k, v of values - if not dv? - dv = allValues.get(path) - if not dv? - dv = {} - allValues.set(path, dv) - dv[k] = v - - if sessionData.revisions[inception] % 2 == 0 - for declaration in decs - aliasDeclarations = device.getAliasDeclarations(declaration[0], declaration[1][0]) - if aliasDeclarations? - for d in aliasDeclarations - path = sessionData.deviceData.paths.add(d[0]) - if path[0] == 'VirtualParameters' or path[0] == '*' - declare(path, d[1], d[2], - sessionData.deviceData.virtualParameterTimestamps, - sessionData.deviceData.virtualParameterValues) - continue if path[0] != '*' - - declare(path, d[1], d[2], - sessionData.deviceData.declarationTimestmaps, - sessionData.deviceData.declarationValues) - else - for declaration in decs - unpacked = device.unpack(sessionData.deviceData, declaration[0]) - - for path in unpacked - path = sessionData.deviceData.paths.add(path) - if path[0] == 'VirtualParameters' or path[0] == '*' - declare(path, declaration[1], declaration[2], - sessionData.deviceData.virtualParameterTimestamps, - sessionData.deviceData.virtualParameterValues) - continue if path[0] != '*' - - declare(path, declaration[1], declaration[2], - sessionData.deviceData.declarationTimestamps, - sessionData.deviceData.declarationValues) - - nextRpc = (applied) -> - ++ sessionData.revisions[inception] - if not applied - if (rpcReq = generateRpcRequest(sessionData))? - sessionData.rpcRequest = rpcReq - return callback(null, generateRpcId(sessionData), rpcReq) - - if sessionData.revisions[inception] % 2 == 0 - return cb() - else - return applyDeclarations(inception, decs, cb) - - if sessionData.revisions.length > inception + 1 - doVirtualParameters(inception + 1, null, nextRpc) - else - loadParameters(sessionData, getParametersToLoad(sessionData), virtualParameters, (err) -> - return callback(err) if err - doVirtualParameters(inception + 1, null, nextRpc) - ) - - applyDeclarations(0, allDeclarations, () -> - return rpcRequest(sessionData, declarations, callback) - ) - ) - - rpcResponse = (sessionData, id, rpcRes, callback) -> if id != generateRpcId(sessionData) return callback(new Error('Request ID not recognized')) @@ -849,23 +868,47 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> 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('*'), - {exist: timestamp}) + root = common.parsePath(rpcReq.parameterPath) + params = [] + params.push([root.concat('*'), {exist: 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.concat('*'), - {exist: timestamp}) + params.push([path.concat('*'), {exist: timestamp}]) - device.set(sessionData.deviceData, path, + params.push([path, {exist: timestamp, object: timestamp, writable: timestamp}, - {exist: 1, object: 1, writable: if p[1] then 1 else 0}) + {exist: 1, object: 1, writable: if p[1] then 1 else 0}]) + else - device.set(sessionData.deviceData, common.parsePath(p[0]), + params.push([common.parsePath(p[0]), {exist: timestamp, object: timestamp, writable: timestamp}, - {exist: 1, object: 0, writable: if p[1] then 1 else 0}) + {exist: 1, object: 0, writable: if p[1] then 1 else 0}]) + + # Sort such that actual parameters are set before wildcard ones + params.sort((a, b) -> + al = a[0].length + bl = b[0].length + -- bl if b[0][bl - 1] == '*' + -- al if a[0][al - 1] == '*' + return bl - al + ) + + p = root.slice() + p.length = params[params.length - 1].length + p.fill('*', root.length) + loadPath(sessionData, p) + loadParameters(sessionData, (err) -> + return callback(err) if err + + for p in params + device.set(sessionData.deviceData, p[0], p[1], p[2]) + + return callback() + ) + return when 'SetParameterValuesResponse' return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'SetParameterValues' @@ -888,35 +931,49 @@ rpcFault = (sessionData, id, faultResponse, callback) -> load = (id, callback) -> db.redisClient.get("session_#{id}", (err, sessionDataString) -> return callback(err) if err or not sessionDataString? - sessionData = JSON.parse(sessionDataString) - deviceData = initDeviceData() - keys = ['exist', 'object', 'writable', 'value'] - for r in sessionData.deviceData - path = deviceData.paths.add(r[0]) - if r[1] - deviceData.loaded.set(path, r[1]) + cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> + return callback(err) if err - for k, i in keys - if r[2 + i]? - deviceData.timestamps[k].setRevisions(path, r[2 + i]) + sessionData = JSON.parse(sessionDataString) - if r[2 + keys.length + i]? - deviceData.values[k].setRevisions(path, r[2 + keys.length + i]) + if sessionData.cache.hash != hash + return callback(new Error('Preset hash mismatch')) - if r[2 + (keys.length * 2) + i]? - deviceData.declarationTimestamps.set(path, r[2 + (keys.length * 2) + i]) + sessionData.cache.provisions = provisions + sessionData.cache.virtualParameters = virtualParameters - if r[2 + (keys.length * 3) + i]? - deviceData.declarationValues.set(path, r[2 + (keys.length * 3) + i]) + for d in sessionData.declarations + common.addPathMeta(d[0]) - sessionData.deviceData = deviceData + deviceData = initDeviceData() + keys = ['exist', 'object', 'writable', 'value'] + for r in sessionData.deviceData + path = deviceData.paths.add(r[0]) - return callback(null, sessionData) + if r[1] + deviceData.loaded.set(path, r[1]) + + for k, i in keys + if r[2 + i]? + deviceData.timestamps[k].setRevisions(path, r[2 + i]) + + if r[2 + keys.length + i]? + deviceData.values[k].setRevisions(path, r[2 + keys.length + i]) + + sessionData.deviceData = deviceData + + return callback(null, sessionData) + ) ) save = (sessionData, callback) -> + delete sessionData.syncState + delete sessionData.toLoad + delete sessionData.cache.provisions + delete sessionData.cache.virtualParameters + sessionData.id ?= crypto.randomBytes(8).toString('hex') deviceData = [] keys = ['exist', 'object', 'writable', 'value'] @@ -932,12 +989,6 @@ save = (sessionData, callback) -> if r = sessionData.deviceData.values[k].getRevisions(path) e[2 + keys.length + i] = r - if r = sessionData.deviceData.declarationTimestamps.get(path) - e[2 + (keys.length * 2) + i] = r - - if r = sessionData.deviceData.declarationValues.get(path) - e[2 + (keys.length * 3) + i] = r - deviceData.push(e) oldDeviceData = sessionData.deviceData.deviceData