diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index c35faf3..3c99f93 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -40,6 +40,7 @@ util = require 'util' zlib = require 'zlib' +mongodb = require 'mongodb' config = require './config' common = require './common' @@ -48,6 +49,7 @@ session = require './session' query = require './query' device = require './device' cache = require './cache' +db = require './db' writeResponse = (currentRequest, res) -> @@ -145,11 +147,51 @@ applyPresets = (currentRequest) -> nextRpc = (currentRequest) -> session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest) -> throw err if err - if not rpcRequest? - session.clearProvisions(currentRequest.sessionData) + if rpcRequest? + return sendRpcRequest(currentRequest, id, rpcRequest) + + doneTasks = null + for p in currentRequest.sessionData.provisions when p[0] == '_task' + doneTasks ?= [] + doneTasks.push(mongodb.ObjectID(p[1])) + + session.clearProvisions(currentRequest.sessionData) + + # No need to query for pending tasks if there was none in previous cycle + if not doneTasks? and currentRequest.sessionData.provisions?.length return applyPresets(currentRequest) - sendRpcRequest(currentRequest, id, rpcRequest) + nextTask = () -> + cur = db.tasksCollection.find({'device' : currentRequest.sessionData.deviceId, timestamp : {$lte : new Date(currentRequest.sessionData.timestamp)}}).sort(['timestamp']).limit(1) + cur.nextObject((err, task) -> + if not task? + return applyPresets(currentRequest) + + switch task.name + when 'getParameterValues' + session.addProvisions(currentRequest.sessionData, [['_task', task._id, 'getParameterValues'].concat(task.parameterNames)]) + when 'setParameterValues' + t = ['_task', task._id, 'setParameterValues'] + for p in task.parameterValues + t.push(p[0]) + t.push(p[1]) + t.push(p[2] ? '') + session.addProvisions(currentRequest.sessionData, [t]) + when 'refreshObject' + session.addProvisions(currentRequest.sessionData, [['_task', task._id, 'refreshObject', task.objectName]]) + else + throw new Error('Task name not recognized') + + return nextRpc(currentRequest) + ) + + if not doneTasks? + return nextTask() + + db.tasksCollection.remove({'_id' : {'$in' : doneTasks}}, (err, res) -> + throw err if err + return nextTask() + ) ) diff --git a/lib/session.coffee b/lib/session.coffee index 45f340b..96cba26 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -364,7 +364,7 @@ rpcRequest = (sessionData, declarations, callback) -> allDeclarations = declarations?.slice() ? [] cache.getProvisionsAndVirtualParameters((err, presetsHash, provisions, virtualParameters) -> - return callback(err) if err or presetsHash != sessionData.presetsHash + return callback(err) if err or (sessionData.presetsHash? and sessionData.presetsHash != presetsHash) done = true _extensions = [] @@ -380,6 +380,21 @@ rpcRequest = (sessionData, declarations, callback) -> allDeclarations.push([common.parsePath(provision[1]), 1, null, null, null, null, null, 1, [provision[2]]]) when 'tag' allDeclarations.push([[['Tags', provision[1]], null, null, null, null, null, null, null, [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]), 1, null, null, null, null, null, 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]), 1, null, null, null, null, null, 1, v]) + when 'refreshObject' + path = common.parsePath(provision[3]) + for i in [path.length...16] by 1 + path.length = i + allDeclarations.push([path.slice(), sessionData.timestamp, null, 1, null, 1, null, sessionData.timestamp]) continue ret = sandbox.run(provisions[provision[0]].script, provision.slice(1), sessionData.deviceData, sessionData.extensionsCache, 0, sessionData.revisions[0] >> 1)