diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index de165c0..be25bef 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -111,13 +111,11 @@ inform = (currentRequest, cwmpRequest) -> informHash = crypto.createHash('md5').update(JSON.stringify(parameterNames)).update(JSON.stringify(config.CUSTOM_COMMANDS)).digest('hex') - actions = {parameterValues : cwmpRequest.methodRequest.parameterList, set : {}} - actions.set._lastInform = now - actions.set._lastBoot = now if '1 BOOT' in cwmpRequest.methodRequest.event + deviceUpdates = {parameterValues : cwmpRequest.methodRequest.parameterList, informEvents : cwmpRequest.methodRequest.event} + # Delete all pending tasks on bootstrap (e.g. factory reset) if '0 BOOTSTRAP' in cwmpRequest.methodRequest.event - actions.set._lastBootstrap = lastBootstrap = now - # ignore all pending tasks + lastBootstrap = now db.tasksCollection.remove({device : currentRequest.session.deviceId}, (err, removed) -> throw err if err ) @@ -126,17 +124,17 @@ inform = (currentRequest, cwmpRequest) -> throw err if err updateAndRespond = () -> - updateDevice(currentRequest.session.deviceId, actions, (err) -> + updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, false, (err) -> throw err if err - res = soap.response({ - id : cwmpRequest.id, - methodResponse : {type : 'InformResponse'}, - cwmpVersion : cwmpRequest.cwmpVersion - }) - res.headers['Set-Cookie'] = "session=#{currentRequest.sessionId}" - writeResponse(currentRequest, res) db.redisClient.setex("session_#{currentRequest.sessionId}", currentRequest.session.sessionTimeout, JSON.stringify(currentRequest.session), (err) -> throw err if err + res = soap.response({ + id : cwmpRequest.id, + methodResponse : {type : 'InformResponse'}, + cwmpVersion : cwmpRequest.cwmpVersion + }) + res.headers['Set-Cookie'] = "session=#{currentRequest.sessionId}" + writeResponse(currentRequest, res) ) ) @@ -189,8 +187,8 @@ inform = (currentRequest, cwmpRequest) -> delete deviceCustomCommands[cmd] for cmd of deviceCustomCommands - actions.deletedObjects ?= [] - actions.deletedObjects.push("_customCommands.#{cmd}") + deviceUpdates.deletedObjects ?= [] + deviceUpdates.deletedObjects.push("_customCommands.#{cmd}") # clear presets hash if parameters are potentially modified if _tasks.length > 0 @@ -227,7 +225,7 @@ runTask = (currentRequest, task, methodResponse) -> tasks.task(task, methodResponse, (err, status, methodRequest, deviceUpdates) -> throw err if err # TODO handle error - updateDevice(currentRequest.session.deviceId, deviceUpdates, (err) -> + updateDevice.queueUpdates(currentRequest.session.deviceId, deviceUpdates, (err) -> throw err if err timeDiff = process.hrtime(timeDiff)[0] + 1 @@ -251,19 +249,25 @@ runTask = (currentRequest, task, methodResponse) -> ) if save - db.tasksCollection.update({_id : mongodb.ObjectID(String(task._id))}, {$set : {session : task.session}}, (err) -> + updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, true, (err) -> throw err if err - f() + db.tasksCollection.update({_id : mongodb.ObjectID(String(task._id))}, {$set : {session : task.session}}, (err) -> + throw err if err + f() + ) ) else f() when tasks.STATUS_COMPLETED util.log("#{currentRequest.session.deviceId}: Completed task #{task.name}(#{task._id})") - db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) -> + updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, true, (err) -> throw err if err - db.redisClient.del(String(task._id), (err, res) -> + db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) -> throw err if err - nextTask(currentRequest) + db.redisClient.del(String(task._id), (err, res) -> + throw err if err + nextTask(currentRequest) + ) ) ) when tasks.STATUS_FAULT @@ -358,7 +362,10 @@ nextTask = (currentRequest) -> ) else util.log("#{currentRequest.session.deviceId}: Started task #{task.name}(#{task._id})") - runTask(currentRequest, task, {}) + updateDevice.clearUpdatesQueue(currentRequest.session.deviceId, (err) -> + throw err if err + runTask(currentRequest, task, {}) + ) ) diff --git a/lib/tasks.coffee b/lib/tasks.coffee index 6a3c919..723e494 100644 --- a/lib/tasks.coffee +++ b/lib/tasks.coffee @@ -65,12 +65,16 @@ this.refreshObject = (task, methodResponse, callback) -> if task.session.subtask.name is 'getParameterNames' this.getParameterNames(task.session.subtask, methodResponse, (err, status, methodRequest, deviceUpdates) => - if status & STATUS_COMPLETED - parameterNames = [] + task.session.parameterNames ?= [] + if deviceUpdates?.parameterNames? for p in deviceUpdates.parameterNames if not common.endsWith(p[0], '.') - parameterNames.push(p[0]) - task.session.subtask = {name : 'getParameterValues', parameterNames : parameterNames} + task.session.parameterNames.push(p[0]) + + if status & STATUS_COMPLETED + task.session.subtask = {name : 'getParameterValues', parameterNames : task.session.parameterNames} + delete task.session.parameterNames + this.getParameterValues(task.session.subtask, {}, (err, status, methodRequest) -> # ignore deviceUpdates returned by firt call to getParameterValues callback(err, status, methodRequest, deviceUpdates) @@ -78,7 +82,7 @@ this.refreshObject = (task, methodResponse, callback) -> else if status & STATUS_OK callback(err, STATUS_OK, methodRequest, deviceUpdates) else - throw Error('Unexpected subtask status') + throw new Error('Unexpected subtask status') ) else if task.session.subtask.name is 'getParameterValues' if methodResponse.faultcode? @@ -103,15 +107,31 @@ this.getParameterNames = (task, methodResponse, callback) -> return (param[...-1] + '.').split('.').length - 1 if methodResponse.type is 'GetParameterNamesResponse' - task.session.parameterNames = task.session.parameterNames.concat(methodResponse.parameterList) path = task.session.queue.pop() # If parameter depth higher than the threshold, nextLevel was set to false if not task.nextLevel? and getParameterDepth(path) < config.GET_PARAMETER_NAMES_DEPTH_THRESHOLD for p in methodResponse.parameterList task.session.queue.push(p[0]) if p[0][-1..] == '.' + + deviceUpdates = {parameterNames : methodResponse.parameterList} + + # Make sure that for each parameter, all its parents are explicitly included + found = {} + found[path] = 0 if !!path + for p in deviceUpdates.parameterNames + param = p[0] + i = param.length + while (i = param.lastIndexOf('.', i-1)) > path.length + pp = param.slice(0, i + 1) + break if found[pp]? + found[pp] = 0 + found[p[0]] = 1 + + for k, v of found + deviceUpdates.parameterNames.push([k]) if v == 0 else - task.session = {queue : [task.parameterPath], parameterNames : []} + task.session = {queue : [task.parameterPath]} if task.session.queue.length > 0 path = task.session.queue[-1..][0] @@ -120,25 +140,8 @@ this.getParameterNames = (task, methodResponse, callback) -> parameterPath : path, nextLevel : task.nextLevel ? getParameterDepth(path) < config.GET_PARAMETER_NAMES_DEPTH_THRESHOLD } - return callback(null, STATUS_OK, methodRequest) + return callback(null, STATUS_OK, methodRequest, deviceUpdates) else - deviceUpdates = {parameterNames : task.session.parameterNames} - - # Make sure that for each parameter, all its parents are explicitly included - found = {} - found[task.parameterPath] = 0 if !!task.parameterPath - for p in deviceUpdates.parameterNames - param = p[0] - i = param.length - while (i = param.lastIndexOf('.', i-1)) > task.parameterPath.length - pp = param.slice(0, i + 1) - break if found[pp]? - found[pp] = 0 - found[p[0]] = 1 - - for k, v of found - deviceUpdates.parameterNames.push([k]) if v == 0 - return callback(null, STATUS_COMPLETED, null, deviceUpdates) diff --git a/lib/update-device.coffee b/lib/update-device.coffee index f37ddb9..c399e10 100644 --- a/lib/update-device.coffee +++ b/lib/update-device.coffee @@ -25,136 +25,197 @@ OBJECT_REGEX = /\.$/ INSTANCE_REGEX = /\.[\d]+\.$/ -updateDevice = (deviceId, actions, callback) -> - return callback?() if not actions? +queueUpdates = (deviceId, deviceUpdates, callback) -> + return callback() if not deviceUpdates? or Object.keys(deviceUpdates).length == 0 + # TODO consider setting expiry + db.redisClient.rpush("#{deviceId}_updates", JSON.stringify(deviceUpdates), (err, res) -> + return callback(err) + ) + +clearUpdatesQueue = (deviceId, callback) -> + db.redisClient.del("#{deviceId}_updates", (err, res) -> + return callback(err) + ) + + +commitUpdates = (deviceId, deviceUpdates, commitQueue, callback) -> now = new Date(Date.now()) update = {'$set' : {}, '$unset' : {}} - if actions.parameterValues? - for p in actions.parameterValues - origValue = p[1] - v = normalize(p[0], origValue) + # Used to find parameters that no longer exist in device + presentPaths = null + projection = null - path = if common.endsWith(p[0], '.') then p[0] else "#{p[0]}." - if v == origValue - update['$unset']["#{path}_orig"] = 1 - else - update['$set']["#{path}_orig"] = origValue - update['$set']["#{path}_value"] = v - update['$set']["#{path}_timestamp"] = now - if p[2]? - update['$set']["#{path}_type"] = p[2] + # Keep track of alias changes to clear cache if needed + newAliases = null - if actions.deletedObjects? - for p in actions.deletedObjects - update['$unset'][p] = 1 + processBatch = (updatesBatch, cb) -> + if updatesBatch.informEvents? + update['$set']['_lastInform'] = now + update['$set']['_lastBoot'] if '1 BOOT' in updatesBatch.informEvents + update['$set']['_lastBootstrap'] if '0 BOOTSTRAP' in updatesBatch.informEvents - if actions.instanceName? - for i in actions.instanceName - update['$set']["#{i[0]}._name"] = i[1] + if updatesBatch.parameterValues? + for p in updatesBatch.parameterValues + origValue = p[1] + v = normalize(p[0], origValue) - if actions.parameterNames? - projection = {} # fetch only necessary parameters - presentPaths = {} # to be compared with to find no longer existing parameters - - for p in actions.parameterNames - if OBJECT_REGEX.test(p[0]) - path = p[0].slice(0, -1) - if INSTANCE_REGEX.test(p[0]) - update['$set']["#{path}._instance"] = true + path = if common.endsWith(p[0], '.') then p[0] else "#{p[0]}." + if v == origValue + update['$unset']["#{path}_orig"] = 1 else - update['$set']["#{path}._object"] = true - else - path = p[0] - update['$set']["#{path}._writable"] = p[1] if p[1]? - update['$set']["#{path}._timestamp"] = now + update['$set']["#{path}_orig"] = origValue + update['$set']["#{path}_value"] = v + update['$set']["#{path}_timestamp"] = now + if p[2]? + update['$set']["#{path}_type"] = p[2] - # Add parameter (sans any dot suffix) if has an alias - if (a = parameters.getAlias(path))? - newAliases = {} if not newAliases? - newAliases[path] = a + if updatesBatch.deletedObjects? + for p in updatesBatch.deletedObjects + update['$unset'][p] = 1 - projection[path] = 1 + if updatesBatch.instanceName? + for i in updatesBatch.instanceName + update['$set']["#{i[0]}._name"] = i[1] - # Store this param path and all its parents for later look up - presentPaths[path] = true - i = p[0].indexOf('.') - while i != -1 - presentPaths[p[0][0...i]] = true - i = p[0].indexOf('.', i + 1) + if updatesBatch.parameterNames? + presentPaths ?= {} + projection ?= {} - mongoQuery.optimizeProjection(projection) + for p in updatesBatch.parameterNames + if OBJECT_REGEX.test(p[0]) + path = p[0].slice(0, -1) + if INSTANCE_REGEX.test(p[0]) + update['$set']["#{path}._instance"] = true + else + update['$set']["#{path}._object"] = true + else + path = p[0] + update['$set']["#{path}._writable"] = p[1] if p[1]? + update['$set']["#{path}._timestamp"] = now - if actions.customCommands? - for p in actions.customCommands - commandName = p[0] - update['$set']["_customCommands.#{commandName}._value"] = p[1] - update['$set']["_customCommands.#{commandName}._timestamp"] = now + # Add parameter (sans any dot suffix) if has an alias + if (a = parameters.getAlias(path))? + newAliases ?= {} + newAliases[path] = a - if actions.set? - common.extend(update['$set'], actions.set) + projection[path] = 1 - f = () -> + # Store this param path and all its parents for later look up + presentPaths[path] = true + i = p[0].indexOf('.') + while i != -1 + presentPaths[p[0][0...i]] = true + i = p[0].indexOf('.', i + 1) + + if updatesBatch.customCommands? + for p in updatesBatch.customCommands + commandName = p[0] + update['$set']["_customCommands.#{commandName}._value"] = p[1] + update['$set']["_customCommands.#{commandName}._timestamp"] = now + + processBatches = (cb) -> + if not commitQueue? + processBatch(deviceUpdates) if deviceUpdates? + return cb() + + db.redisClient.lpop("#{deviceId}_updates", (err, res) -> + callback(err) if err + + if not res? + processBatch(deviceUpdates) if deviceUpdates? + return cb() + + processBatch(JSON.parse(res)) + return processBatches(cb) + ) + + commit = () -> for k of update delete update[k] if Object.keys(update[k]).length == 0 if Object.keys(update).length db.devicesCollection.update({'_id' : deviceId}, update, {}, (err, count) -> - throw err if err + return callback(err) if err + # Clear aliases cache if there's a new aliased parameter that has not been included in aliases cache if newAliases? + newAliasFound = false db.getAliases((aliases) -> for p, a of newAliases if not aliases[a]? or p not in aliases[a] - db.redisClient.del('aliases', (err, res) -> - throw err if err - ) + newAliasFound = true break + + return callback() if not newAliasFound + db.redisClient.del('aliases', (err, res) -> + return callback(err) + ) ) - callback?(err) + else + return callback() ) else - callback?() + return callback() - if projection? - db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> - throw err if err + processBatches(() -> + if projection? + mongoQuery.optimizeProjection(projection) - recursive = (obj, prefix) -> - for k,v of obj - p = prefix + k - if k[0] == '_' - # Remove set operations for unmodified values for optimization - if update['$set'][p] == v - delete update['$set'][p] - continue + # Fetch device from DB to discover and delete parameters that no longer exist + # While we're fetching the object, we might as well use the information + # to optimize our update query + db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> + return callback(err) if err - # Remove unset operations for values that don't exist - delete update['$unset']["#{p}._orig"] if not v._orig? + recursive = (obj, prefix) -> + for k, v of obj + p = prefix + k - if presentPaths[p] - recursive(v, "#{p}.") + # Mark unset operations we want to keep + if update['$unset'][p]? + update['$unset'][p] = 2 + continue + + if k[0] == '_' + # Remove set operations for unmodified values for optimization + if update['$set'][p] == v + delete update['$set'][p] + continue + + if presentPaths[p] + recursive(v, "#{p}.") + else + update['$unset'][p] = 1 + + # Scan the DB object and remove parameters that no longer exists in device + for path of projection + root = device + rootPath = '' + for p in path.split('.') + rootPath += "#{p}." + root = root[p] + break if not root? + + if root + recursive(root, rootPath) else - update['$unset'][p] = 1 + if not presentPaths[rootPath.slice(0, -1)]? + update['$unset'][rootPath.slice(0, -1)] = 1 - # Scan the DB object and remove parameters that no longer exists in device - for k, v of projection - root = device - rootPath = '' - for p in path.split('.') - rootPath += "#{p}." - root = root[p] - break if not root? + # Remove unnecessary unset updates + for k, v of update['$unset'] + if common.startsWith(k, path + '.') and v != 2 + delete update['$unset'][k] - if root - recursive(root, rootPath) - else - if not presentPaths[rootPath.slice(0, -1)] - update['$unset'][rootPath.slice(0, -1)] = 1 - f() - ) - else - f() + commit() + ) + else + commit() + ) -module.exports = updateDevice \ No newline at end of file + +exports.queueUpdates = queueUpdates +exports.clearUpdatesQueue = clearUpdatesQueue +exports.commitUpdates = commitUpdates