### # Copyright 2013-2016 Zaid Abdulla # # GenieACS is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # GenieACS is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with GenieACS. If not, see . ### config = require './config' common = require './common' db = require './db' device = require './device' sandbox = require './sandbox' cache = require './cache' extensions = require './extensions' PathSet = require './path-set' VersionedMap = require './versioned-map.js' InstanceSet = require './instance-set' initDeviceData = () -> return { 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()} loaded : new Map() } init = (deviceId, cwmpVersion, timeout, callback) -> timestamp = Date.now() sessionData = { timestamp : timestamp deviceId : deviceId deviceData : initDeviceData() cwmpVersion : cwmpVersion timeout : timeout provisions: [[]] channels: [] revisions: [0] rpcCount: 0 cycle: 0 extensionsCache: [] declarations: [] } 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, callback) -> if not sessionData.toLoad?.length return callback() 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 loaded = [[[], 99]] parameters = [] vparams = {} for p in loaded path = sessionData.deviceData.paths.add(p[0]) if p[1] l = sessionData.deviceData.loaded.get(path) || 0 sessionData.deviceData.loaded.set(path, Math.max(l, p[1])) if path.length <= 2 and path.length + p[1] > 2 if path.length < 1 or path[0] == '*' or path[0] == 'VirtualParameters' if path.length < 2 or path[1] == '*' vparams['*'] = 1 else vparams[path[1]] = 1 if path.length <= 1 and path.length + p[1] > 1 if path.length < 1 or path[0] == '*' or path[0] == 'VirtualParameters' vparams[''] = 1 for p in parameters path = sessionData.deviceData.paths.add(p[0]) for k, v of p[1] sessionData.deviceData.timestamps[k].set(path, v, 0) for k, v of p[2] sessionData.deviceData.values[k].set(path, v, 0) # Virtual parameters if vparams[''] device.set(sessionData.deviceData, ['VirtualParameters'], {exist: sessionData.timestamp, object: sessionData.timestamp, writable: sessionData.timestamp}, {exist: 1, object: 1, writable: 0}) delete vparams[''] if vparams['*'] for k, v of sessionData.cache.virtualParameters device.set(sessionData.deviceData, ['VirtualParameters', k], {exist: sessionData.timestamp, object: sessionData.timestamp}, {exist: 1, object: 0}) device.set(sessionData.deviceData, ['VirtualParameters', '*'], {exist: sessionData.timestamp}) else for k of vparams if k of sessionData.cache.virtualParameters device.set(sessionData.deviceData, ['VirtualParameters', k] {exist: sessionData.timestamp, object: sessionData.timestamp}, {exist: 1, object: 0}) else device.set(sessionData.deviceData, ['VirtualParameters', k], {exist: sessionData.timestamp}) delete sessionData.toLoad return callback() ) generateRpcId = (sessionData) -> if sessionData.rpcCount > 255 or sessionData.cycle > 15 or sessionData.revisions.length > 15 throw new Error('Too many RPCs') return sessionData.timestamp.toString(16) + "0#{sessionData.rpcCount.toString(16)}".slice(-2) inform = (sessionData, rpcReq, callback) -> timestamp = sessionData.timestamp params = [] params.push([['DeviceID', 'Manufacturer'], {exist: timestamp, object: timestamp, writable: timestamp, value: timestamp}, {exist: 1, object: 0, writable: 0, value: [rpcReq.deviceId.Manufacturer, 'xsd:string']}]) params.push([['DeviceID', 'OUI'], {exist: timestamp, object: timestamp, writable: timestamp, value: timestamp}, {exist: 1, object: 0, writable: 0, value: [rpcReq.deviceId.OUI, 'xsd:string']}]) params.push([['DeviceID', 'ProductClass'], {exist: timestamp, object: timestamp, writable: timestamp, value: timestamp}, {exist: 1, object: 0, writable: 0, value: [rpcReq.deviceId.ProductClass, 'xsd:string']}]) params.push([['DeviceID', 'SerialNumber'], {exist: timestamp, object: timestamp, writable: timestamp, value: timestamp}, {exist: 1, object: 0, writable: 0, value: [rpcReq.deviceId.SerialNumber, 'xsd:string']}]) for p in rpcReq.parameterList path = common.parsePath(p[0]) params.push([path, {exist: timestamp, object: timestamp, value: timestamp}, {exist: 1, object: 0, value: p.slice(1)}]) params.push([['Events', 'Inform'], {exist: timestamp, object: timestamp, writable: timestamp, value: timestamp}, {exist: 1, object: 0, writable: 0, value: [timestamp, 'xsd:dateTime']}]) for e in rpcReq.event params.push([['Events', e.replace(' ', '_')], {exist: timestamp, object: timestamp, writable: timestamp, value: timestamp}, {exist: 1, object: 0, writable: 0, value: [timestamp, 'xsd:dateTime']}]) for p in params loadPath(sessionData, p[0]) loadParameters(sessionData, (err) -> return callback(err) if err v.revision = 1 for k, v of sessionData.deviceData.timestamps v.revision = 1 for k, v of sessionData.deviceData.values for p in params device.set(sessionData.deviceData, p[0], p[1], p[2]) return callback(null, {type : 'InformResponse'}) ) addProvisions = (sessionData, channel, provisions) -> sessionData.declarations = [] sessionData.provisions.length = 1 sessionData.doneProvisions = 0 sessionData.doneDeclarations = 0 for provision, i in provisions # Remove duplicate provisions for p, j in sessionData.provisions[0] if channel == sessionData.channels[j] and JSON.stringify(p) == JSON.stringify(provision) sessionData.provisions[0].splice(j, 1) sessionData.channels.splice(j, 1) 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.timestamps v.collapse(1) for k, v of sessionData.values sessionData.cycle += 1 sessionData.rpcCount = 0 sessionData.revisions = [0] sessionData.extensionsCache.length = 0 clearProvisions = (sessionData) -> delete sessionData.syncState sessionData.provisions = [[]] sessionData.channels = [] sessionData.declarations = [] sessionData.doneProvisions = 0 sessionData.doneDeclarations = 0 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.values sessionData.cycle += 1 sessionData.rpcCount = 0 sessionData.revisions = [0] sessionData.extensionsCache.length = 0 runProvisions = (sessionData, provisions, startRevision, endRevision, callback) -> done = true _extensions = [] allDeclarations = [] for provision in 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 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}]) when 'reboot' allDeclarations.push([['Reboot'], {exist: 1, value: 1}, {value: [sessionData.timestamp]}]) when 'factoryReset' allDeclarations.push([['FactoryReset'], {exist: 1, value: 1}, {value: [sessionData.timestamp]}]) continue ret = sandbox.run(sessionData.cache.provisions[provision[0]].script, {args: provision[1]}, sessionData.timestamp, sessionData.deviceData, sessionData.extensionsCache, startRevision, endRevision) if ret.extensions?.length _extensions = _extensions.concat(ret.extensions) done &&= ret.done allDeclarations = allDeclarations.concat(ret.declarations) if _extensions.length return runExtensions(sessionData, endRevision, _extensions, (err) -> return callback(err) if err return runProvisions(sessionData, provisions, startRevision, endRevision, callback) ) return callback(null, done, allDeclarations) runVirtualParameters = (sessionData, provisions, startRevision, endRevision, callback) -> done = true decs = [] virtualParameterUpdates = [] _extensions = [] for provision in provisions ret = sandbox.run(sessionData.cache.virtualParameters[provision[0]].script, provision[1], sessionData.timestamp, sessionData.deviceData, sessionData.extensionsCache, startRevision, endRevision) if ret.extensions?.length _extensions = _extensions.concat(ret.extensions) done &&= ret.done decs = decs.concat(ret.declarations) if ret.done virtualParameterUpdates.push([provision[0], ret.returnValue]) if _extensions.length return runExtensions(sessionData, endRevision, _extensions, (err) -> return callback(err) if err return runVirtualParameters(sessionData, provisions, startRevision, endRevision, callback) ) if virtualParameterUpdates.length == provisions.length for vpu, i in virtualParameterUpdates commitVirtualParameter(sessionData, vpu[0], provisions[i][1], vpu[1]) return callback(null, done, decs) runDeclarations = (sessionData, declarations, callback) -> done = true if sessionData.toLoad?.length return loadParameters(sessionData, (err) -> return callback(err) if err runDeclarations(sessionData, declarations, callback) ) for d in declarations done &= applyDeclaration(sessionData, d[0], d[1], d[2]) if sessionData.syncState? and sessionData.syncState.virtualParameters? provisions = [] for k, v of sessionData.syncState.virtualParameters provisions.push([k, v]) sessionData.syncState.virtualParameters = {} return callback(null, done, provisions) rpcRequest = (sessionData, _declarations, callback) -> if sessionData.rpcRequest? return callback(null, generateRpcId(sessionData), sessionData.rpcRequest) if sessionData.syncState? and (rpcReq = generateRpcRequest(sessionData))? sessionData.rpcRequest = rpcReq return callback(null, generateRpcId(sessionData), rpcReq) revision = sessionData.revisions[sessionData.revisions.length - 1] + 1 v.revision = revision for k, v of sessionData.deviceData.timestamps v.revision = revision for k, v of sessionData.deviceData.values if _declarations?.length sessionData.declarations = [] sessionData.provisions[0] ?= [] sessionData.revisions[0] ?= 0 sessionData.doneDeclarations = 0 sessionData.doneProvisions = 0 if sessionData.declarations.length != sessionData.provisions.length inception = sessionData.declarations.length run = if inception == 0 then runProvisions else runVirtualParameters return run(sessionData, sessionData.provisions[inception], sessionData.revisions[inception - 1] ? 0, sessionData.revisions[inception], f = (err, done, decs) -> return callback(err) if err if _declarations?.length decs = decs.concat(_declarations) sessionData.declarations.push(decs) sessionData.doneProvisions |= 1 << inception if done if ++ inception < sessionData.provisions.length if inception == 1 and _declarations?.length for d in _declarations sessionData.declarations[0].push(d) for ad in device.getAliasDeclarations(d[0], 1) loadPath(sessionData, ad[0]) else if not decs.length while sessionData.doneDeclarations & (1 << (sessionData.declarations.length - 1)) sessionData.declarations.pop() while sessionData.provisions.length > (sessionData.declarations.length or 1) and sessionData.doneProvisions & (1 << (sessionData.provisions.length - 1)) sessionData.provisions.pop() sessionData.revisions.pop() rev = sessionData.revisions[sessionData.revisions.length - 1] m.collapse(rev + 1) for k, m of sessionData.deviceData.timestamps m.collapse(rev + 1) for k, m of sessionData.deviceData.values sessionData.extensionsCache.splice(rev) sessionData.doneDeclarations &= (1 << sessionData.declarations.length) - 1 sessionData.doneProvisions &= (1 << sessionData.provisions.length) - 1 if sessionData.provisions.length == 1 and sessionData.doneProvisions return callback() for d in decs for ad in device.getAliasDeclarations(d[0], 1) loadPath(sessionData, ad[0]) return rpcRequest(sessionData, null, callback) return runVirtualParameters(sessionData, sessionData.provisions[inception], sessionData.revisions[inception - 1], sessionData.revisions[inception], f) ) return callback() if not sessionData.declarations.length inception = 0 runDeclarations(sessionData, sessionData.declarations[inception], f = (err, done, provisions) -> return callback(err) if err sessionData.doneDeclarations |= 1 << (sessionData.revisions.length - 1) if ++ inception < sessionData.revisions.length return runDeclarations(sessionData, sessionData.declarations[inception], f) if provisions?.length rev = sessionData.revisions[sessionData.revisions.length - 1] sessionData.revisions.push(rev) sessionData.provisions.push(provisions) return runVirtualParameters(sessionData, provisions, rev, rev, (err, done, declarations) -> return callback(err) if err sessionData.doneProvisions |= 1 << (sessionData.revisions.length - 1) if done for d in declarations for ad in device.getAliasDeclarations(d[0], 1) loadPath(sessionData, ad[0]) sessionData.declarations.push(declarations) runDeclarations(sessionData, declarations, f) ) rpcReq = generateRpcRequest(sessionData) if not rpcReq? while sessionData.doneDeclarations & (1 << (sessionData.declarations.length - 1)) sessionData.declarations.pop() while sessionData.provisions.length > (sessionData.declarations.length or 1) and sessionData.doneProvisions & (1 << (sessionData.provisions.length - 1)) sessionData.provisions.pop() sessionData.revisions.pop() rev = sessionData.revisions[sessionData.revisions.length - 1] m.collapse(rev + 1) for k, m of sessionData.deviceData.timestamps m.collapse(rev + 1) for k, m of sessionData.deviceData.values sessionData.extensionsCache.splice(rev) sessionData.doneDeclarations &= (1 << sessionData.declarations.length) - 1 sessionData.doneProvisions &= (1 << sessionData.provisions.length) - 1 if sessionData.provisions.length == 1 and sessionData.doneProvisions return callback() ++ sessionData.revisions[sessionData.revisions.length - 1] return rpcRequest(sessionData, _declarations, callback) sessionData.rpcRequest = rpcReq return callback(null, generateRpcId(sessionData), rpcReq) ) generateRpcRequest = (sessionData) -> if not (syncState = sessionData.syncState)? return iter = syncState.refreshAttributes.exist.values() while (path = iter.next().value) descendantIter = syncState.paths.subset(path, 99) found = false while (p = descendantIter.next().value)? 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 = syncState.paths.add(path.slice(0, -1)) syncState.refreshGpn.set(path, syncState.refreshGpn.get(path) or 1) syncState.refreshAttributes.exist.clear() iter = syncState.refreshAttributes.object.values() while (path = iter.next().value) descendantIter = syncState.paths.subset(path, 99) found = false while (p = descendantIter.next().value)? if syncState.refreshAttributes.value.has(p) or (p.length > path.length and (syncState.refreshAttributes.object.has(p) or syncState.refreshAttributes.writable.has(p))) found = true break if not found path = syncState.paths.add(path.slice(0, -1)) syncState.refreshGpn.set(path, syncState.refreshGpn.get(path) or 1) syncState.refreshAttributes.object.clear() iter = syncState.refreshAttributes.writable.values() while (path = iter.next().value) path = syncState.paths.add(path.slice(0, -1)) syncState.refreshGpn.set(path, syncState.refreshGpn.get(path) or 1) syncState.refreshAttributes.writable.clear() iter = syncState.refreshGpn.keys() while (path = iter.next().value) descendantIter = syncState.paths.subset(path, 99) found = false while (p = descendantIter.next().value)? continue if p == path if (flags = syncState.refreshGpn.get(p)) flags <<= p.length - path.length syncState.refreshGpn.set(path, syncState.refreshGpn.get(path) | flags) syncState.refreshGpn.delete(p) rpcReq = null completed = true if syncState.refreshGpn.size if rpcReq? completed = false else GET_PARAMETER_NAMES_DEPTH_THRESHOLD = config.get('GET_PARAMETER_NAMES_DEPTH_THRESHOLD', sessionData.deviceId) pair = syncState.refreshGpn.entries().next().value nextLevel = true if pair[0].length >= GET_PARAMETER_NAMES_DEPTH_THRESHOLD if common.hammingWeight(pair[1] >> 3) >= 3 nextLevel = false if syncState.refreshGpn.size > 1 or (pair[1] > 1 and nextLevel) completed = false rpcReq = { type: 'GetParameterNames' parameterPath: pair[0].concat('').join('.') nextLevel: nextLevel } syncState.refreshGpn.delete(pair[0]) if syncState.refreshAttributes.value.size if rpcReq? completed = false else TASK_PARAMETERS_BATCH_SIZE = config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId) parameterNames = [] iter = syncState.refreshAttributes.value.values() while (path = iter.next().value) and parameterNames.length < TASK_PARAMETERS_BATCH_SIZE parameterNames.push(path) syncState.refreshAttributes.value.delete(path) if syncState.refreshAttributes.value.size > parameterNames.length completed = false rpcReq = { type: 'GetParameterValues' parameterNames: (p.join('.') for p in parameterNames) } refreshDone = not rpcReq? # Delete instance syncState.instancesToDelete.forEach((instances, parent) -> instances.forEach((instance, key, thisSet) -> if rpcReq? completed = false else rpcReq = { type: 'DeleteObject' objectName: instance.join('.') + '.' } thisSet.delete(instance) ) ) syncState.instancesToDelete.clear() if not refreshDone # Create instance syncState.instancesToCreate.forEach((instances, parent) -> instances.forEach((instance, key, thisSet) -> if rpcReq? completed = false else rpcReq = { type: 'AddObject' objectName: parent.join('.') + '.' next: 'getInstanceKeys' instanceValues: instance } thisSet.delete(instance) ) ) syncState.instancesToCreate.clear() if not refreshDone # Set values TASK_PARAMETERS_BATCH_SIZE = config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId) parameterValues = [] syncState.setValues.forEach((v, k) -> return if completed == false if (curVal = sessionData.deviceData.values.value.get(k))? and sessionData.deviceData.values.writable.get(k) val = v.slice() val[1] = curVal[1] if not val[1]? device.sanitizeParameterValue(val) if val[0] != curVal[0] or val[1] != curVal[1] if rpcReq? or parameterValues.length >= TASK_PARAMETERS_BATCH_SIZE completed = false return parameterValues.push([k, val[0], val[1]]) syncState.setValues.delete(k) ) syncState.setValues.clear() if not refreshDone if parameterValues.length and not rpcReq? rpcReq = { type: 'SetParameterValues' parameterList: ([p[0].join('.'), p[1], p[2]] for p in parameterValues) } # Reboot if syncState.reboot? p = sessionData.deviceData.paths.subset(['Reboot']).next().value if not (p? and sessionData.deviceData.values.value.get(p)?[0] >= syncState.reboot) if rpcReq? completed = false else delete syncState.reboot rpcReq = { type: 'Reboot' } delete syncState.reboot if not refreshDone # Factory reset if syncState.factoryReset? p = sessionData.deviceData.paths.subset(['FactoryReset']).next().value if not (p? and sessionData.deviceData.values.value.get(p)?[0] >= syncState.factoryReset) if rpcReq? completed = false else delete syncState.factoryReset rpcReq = { type: 'FactoryReset' } delete syncState.factoryReset if not refreshDone if completed and rpcReq? rpcReq.done = true return rpcReq 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 sub[0] isnt 'VirtualParameters' continue if not deviceData.values.exist.get(sub) isWritable = deviceData.values.writable.get(sub) 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 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 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 return ret processRebootDeclaration = (sessionData, path, timestamps, values) -> return true if path.length != 1 if values?.value? sessionData.syncState.reboot = values.value[0] return true processFactoryResetDeclaration = (sessionData, path, timestamps, values) -> return true if path.length != 1 if values?.value? sessionData.syncState.factoryReset = values.value[0] return true processDeclaration = (sessionData, path, _timestamps, values) -> syncState = sessionData.syncState deviceData = sessionData.deviceData timestamps = {} timestamps[k] = v for k, v of _timestamps timestamps.exist ?= 1 ret = true if path[0] == '*' or path[0] == 'VirtualParameters' ret = processVirtualparameterDeclaration(sessionData, path, _timestamps, values) && ret return ret if path[0] != '*' if path[0] == '*' or path[0] == 'Reboot' ret = processRebootDeclaration(sessionData, path, _timestamps, values) && ret return ret if path[0] != '*' if path[0] == '*' or path[0] == 'FactoryReset' ret = processFactoryResetDeclaration(sessionData, path, _timestamps, values) && ret 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? syncState.setValues.set(param, values.value) 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) isObject = deviceData.values.object.get(sub) isWritable = deviceData.values.writable.get(sub) 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? syncState.setValues.set(sub, values.value) 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 ret 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) && ret 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) -> return true if sessionData.new 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 processInstances = (sessionData, parent, parameters, keys, minInstances, maxInstances) -> instancesToDelete = sessionData.syncState.instancesToDelete.get(parent) if not instancesToDelete? instancesToDelete = new Set() sessionData.syncState.instancesToDelete.set(parent, instancesToDelete) instancesToCreate = sessionData.syncState.instancesToCreate.get(parent) if not instancesToCreate? instancesToCreate = new InstanceSet() sessionData.syncState.instancesToCreate.set(parent, instancesToCreate) counter = 0 for p in parameters ++ counter if counter > maxInstances instancesToDelete.add(p) else if counter <= minInstances instancesToDelete.delete(p) superset = instancesToCreate.superset(keys) for inst in superset ++ counter if counter > maxInstances instancesToCreate.delete(inst) subset = instancesToCreate.subset(keys) for inst in subset ++ counter if counter <= minInstances instancesToCreate.delete(inst) instancesToCreate.add(JSON.parse(JSON.stringify(keys))) while counter < minInstances ++ counter instancesToCreate.add(JSON.parse(JSON.stringify(keys))) return applyDeclaration = (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: {} instancesToDelete: new Map() instancesToCreate: new Map() } decs = device.getAliasDeclarations(path, timestamps.exist ? 1) if decs.length == 1 processDeclaration(sessionData, decs[0][0], timestamps, values) return true satisfied = true for d in decs satisfied = processDeclaration(sessionData, d[0], d[1]) && satisfied return false if not satisfied unpacked = device.unpack(sessionData.deviceData, path) if values.exist? if Array.isArray(values.exist) minInstances = values.exist[0] maxInstances = values.exist[1] else minInstances = maxInstances = values.exist parent = path.slice(0, -1) if Array.isArray(path[path.length - 1]) keys = {} for p, i in path[path.length - 1] by 2 keys[p] = path[path.length - 1][i + 1] else if path[path.length - 1] == '*' keys = {} if (path.wildcard | path.alias) & ((1 << (path.length - 1)) - 1) == 0 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) for up in unpacked processDeclaration(sessionData, path, timestamps, values) return true commitVirtualParameter = (sessionData, name, declaration, update) -> t = {} v = {} if update.writable? if update.writable[0] <= 0 t.writable = sessionData.timestamp + update.writable[0] else t.writable = Math.min(sessionData.timestamp, update.writable[0]) if declaration.timestamps.writable? t.writable = Math.max(declaration.timestamps.writable, t.writable) v.writable = +update.writable[1] else if declaration.timestamps.writable? throw new Error('Virtual parameter must provide declared attributes') if update.value? if update.value[0] <= 0 t.value = sessionData.timestamp + update.value[0] else t.value = Math.min(sessionData.timestamp, update.value[0]) if declaration.timestamps.value? t.value = Math.max(declaration.timestamps.value, t.value) v.value = device.sanitizeParameterValue(update.value[1]) else if declaration.timestamps.value? throw new Error('Virtual parameter must provide declared attributes') device.set(sessionData.deviceData, ['VirtualParameters', name], t, v) runExtensions = (sessionData, revision, _extensions, callback) -> sessionData.extensionsCache[revision] ?= {} obj = {} for e in _extensions obj[JSON.stringify(e)] = e counter = 1 for k, v of obj ++ counter do (k, v) -> extensions.run(v, (err, res) -> sessionData.extensionsCache[revision][k] = res if -- counter == 0 or err counter = 0 return callback(err) ) if -- counter == 0 return callback() rpcResponse = (sessionData, id, rpcRes, callback) -> if id != generateRpcId(sessionData) return callback(new Error('Request ID not recognized')) ++ sessionData.rpcCount rpcReq = sessionData.rpcRequest if not rpcReq.next? sessionData.rpcRequest = null else if rpcReq.next == 'getInstanceKeys' instanceNumber = rpcRes.instanceNumber parameterNames = [] instanceValues = {} for k, v of rpcReq.instanceValues n = "#{rpcReq.objectName}#{rpcRes.instanceNumber}.#{k}" parameterNames.push(n) instanceValues[n] = v if parameterNames.length == 0 sessionData.rpcRequest = null else sessionData.rpcRequest = { type: 'GetParameterValues' parameterNames: parameterNames next: 'setInstanceKeys' instanceValues: instanceValues } else if rpcReq.next == 'setInstanceKeys' parameterList = [] for p in rpcRes.parameterList if p[1] != rpcReq.instanceValues[p[0]] parameterList.push( [p[0]].concat(device.sanitizeParameterValue([rpcReq.instanceValues[p[0]], p[2]]))) if not parameterList.length sessionData.rpcRequest = null else sessionData.rpcRequest = { type: 'SetParameterValues' parameterList: parameterList } timestamp = sessionData.timestamp 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 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]), {exist: timestamp, object: timestamp, value: timestamp}, {exist: 1, object: 0, value: p.slice(1)}) when 'GetParameterNamesResponse' return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'GetParameterNames' if rpcReq.parameterPath.endsWith('.') root = common.parsePath(rpcReq.parameterPath.slice(0, -1)) else 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 params.push([path.concat('*'), {exist: timestamp}]) params.push([path, {exist: timestamp, object: timestamp, writable: timestamp}, {exist: 1, object: 1, writable: if p[1] then 1 else 0}]) else params.push([common.parsePath(p[0]), {exist: timestamp, object: timestamp, writable: timestamp}, {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 ) for p in params loadPath(sessionData, p[0]) 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' for p in rpcReq.parameterList device.set(sessionData.deviceData, common.parsePath(p[0]), {exist: timestamp, object: timestamp, writable: timestamp, value: timestamp}, {exist: 1, object: 0, writable: 1, value: p.slice(1)}) when 'AddObjectResponse' device.set(sessionData.deviceData, common.parsePath(rpcReq.objectName + rpcRes.instanceNumber), {exist: timestamp, object: timestamp}, {exist: 1, object: 1}) when 'DeleteObjectResponse' device.set(sessionData.deviceData, common.parsePath(rpcReq.objectName.slice(0, -1)), {exist: timestamp}) when 'RebootResponse' device.set(sessionData.deviceData, common.parsePath('Reboot'), {exist: sessionData.timestamp, value: sessionData.timestamp}, {exist: 1, value: [sessionData.timestamp, 'xsd:unsignedInt']}) when 'FactoryResetResponse' device.set(sessionData.deviceData, common.parsePath('FactoryReset'), {exist: sessionData.timestamp, value: sessionData.timestamp}, {exist: 1, value: [sessionData.timestamp, 'xsd:unsignedInt']}) else return callback(new Error('Response type not recognized')) return callback() rpcFault = (sessionData, id, faultResponse, callback) -> # TODO Consider handling invalid parameter faults by automatically refreshing # relevant data model portions delete sessionData.rpcRequest return callback(null, faultResponse) deserialize = (sessionDataString, callback) -> cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> return callback(err) if err sessionData = JSON.parse(sessionDataString) if sessionData.presetsHash? and sessionData.presetsHash != hash return callback(new Error('Preset hash mismatch')) sessionData.cache = { provisions: provisions virtualParameters: virtualParameters } for d in sessionData.declarations common.addPathMeta(d[0]) 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]) 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) ) serialize = (sessionData, callback) -> deviceData = [] keys = ['exist', 'object', 'writable', 'value'] iter = sessionData.deviceData.paths.find([], 99) while not (p = iter.next()).done path = p.value e = [path] e[1] = sessionData.deviceData.loaded.get(path) || 0 for k, i in keys if r = sessionData.deviceData.timestamps[k].getRevisions(path) e[2 + i] = r if r = sessionData.deviceData.values[k].getRevisions(path) e[2 + keys.length + i] = r deviceData.push(e) oldDeviceData = sessionData.deviceData oldSyncState = sessionData.syncState oldToLoad = sessionData.toLoad oldCache = sessionData.cache sessionData.deviceData = deviceData delete sessionData.syncState delete sessionData.toLoad delete sessionData.cache sessionDataString = JSON.stringify(sessionData) sessionData.deviceData = oldDeviceData sessionData.syncState = oldSyncState sessionData.toLoad = oldToLoad sessionData.cache = oldCache return callback(null, sessionDataString) end = (sessionData, callback) -> db.saveDevice(sessionData.deviceId, sessionData.deviceData, sessionData.new, (err) -> return callback(err) if err db.redisClient.del("session_#{sessionData.id}", (err) -> callback(err, sessionData.new) ) ) exports.init = init exports.inform = inform exports.addProvisions = addProvisions exports.clearProvisions = clearProvisions exports.rpcRequest = rpcRequest exports.rpcResponse = rpcResponse exports.rpcFault = rpcFault exports.end = end exports.serialize = serialize exports.deserialize = deserialize