diff --git a/acs.coffee b/acs.coffee index 1d87c43..66da7e5 100644 --- a/acs.coffee +++ b/acs.coffee @@ -93,27 +93,39 @@ updateDevice = (currentRequest, actions, callback) -> runTask = (currentRequest, task, methodResponse) -> cwmpResponse = {} - actions = tasks.task(task, methodResponse, cwmpResponse) + deviceUpdates = {} + status = tasks.task(task, methodResponse, cwmpResponse, deviceUpdates) - updateDevice(currentRequest, actions) + updateDevice(currentRequest, deviceUpdates) - if Object.keys(cwmpResponse).length > 0 - db.updateTask(task, (err) -> - res = tr069.response(task._id, cwmpResponse) - writeResponse(currentRequest, res) - ) - return - - # Task finished - db.memcached.del(String(task._id)) - db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) -> - util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})") - cookies = {} - nextTask(currentRequest, cookies) - ) + switch status + when tasks.STATUS_FINISHED + db.memcached.del(String(task._id)) + db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) -> + util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})") + nextTask(currentRequest) + ) + when tasks.STATUS_FAULT + db.saveTask(task, (err) -> + # Faulty task. No more work to do until task is deleted. + res = tr069.response(null, cwmpResponse) + writeResponse(currentRequest, res) + ) + when tasks.STATUS_PENDING + db.saveTask(task, (err) -> + # task expects CPE confirmation later + nextTask(currentRequest) + ) + when tasks.STATUS_STARTED + db.updateTask(task, (err) -> + res = tr069.response(task._id, cwmpResponse) + writeResponse(currentRequest, res) + ) + else + throw Error('Unknown task status') -nextTask = (currentRequest, cookies) -> +nextTask = (currentRequest) -> cur = db.tasksCollection.find({'device' : currentRequest.deviceId}).sort(['timestamp']).limit(1) cur.nextObject( (err, task) -> cwmpResponse = {} @@ -136,18 +148,17 @@ nextTask = (currentRequest, cookies) -> ) else # no discrepancy, return empty response - res = tr069.response(null, cwmpResponse, cookies) + res = tr069.response(null, cwmpResponse) writeResponse(currentRequest, res) ) else # no discrepancy, return empty response - res = tr069.response(null, cwmpResponse, cookies) + res = tr069.response(null, cwmpResponse) writeResponse(currentRequest, res) ) - return else if task.fault? # last task was faulty. Do nothing until until task is deleted - res = tr069.response(null, cwmpResponse, cookies) + res = tr069.response(null, cwmpResponse) writeResponse(currentRequest, res) else util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})") @@ -244,21 +255,23 @@ else db.getTask(taskId, (task) -> if not task - nextTask(currentRequest, cookies) + nextTask(currentRequest) else runTask(currentRequest, task, cwmpRequest.methodResponse) ) else if cwmpRequest.fault? taskId = cwmpRequest.id util.log("#{currentRequest.deviceId}: Fault response for task #{taskId}") - db.tasksCollection.update({_id : mongodb.ObjectID(taskId)}, {$set : {fault : cwmpRequest.fault}}, (err) -> - # Faulty task. No more work to do until task is deleted. - res = tr069.response(null, cwmpResponse, cookies) - writeResponse(currentRequest, res) + + db.getTask(taskId, (task) -> + if not task + nextTask(currentRequest) + else + runTask(currentRequest, task, cwmpRequest.fault) ) else # cpe sent empty response. start sending acs requests - nextTask(currentRequest, cookies) + nextTask(currentRequest) server.listen config.ACS_PORT, config.ACS_INTERFACE diff --git a/db.coffee b/db.coffee index f4b4199..0285ec8 100644 --- a/db.coffee +++ b/db.coffee @@ -56,6 +56,14 @@ updateTask = (task, callback) -> ) +saveTask = (task, callback) -> + task._id = mongodb.ObjectID(String(task._id)) + tasksCollection.save(task, (err) -> + callback(err) + ) + + exports.memcached = memcached exports.getTask = getTask exports.updateTask = updateTask +exports.saveTask = saveTask diff --git a/tasks.coffee b/tasks.coffee index b271906..425681d 100644 --- a/tasks.coffee +++ b/tasks.coffee @@ -1,37 +1,64 @@ common = require './common' BATCH_SIZE = 16 -this.init = (task, methodResponse, cwmpResponse) -> - ret = {} +exports.STATUS_QUEUED = STATUS_QUEUED = 0 +exports.STATUS_STARTED = STATUS_STARTED = 1 +exports.STATUS_PENDING = STATUS_PENDING = 2 +exports.STATUS_FAULT = STATUS_FAULT = 3 +exports.STATUS_FINISHED = STATUS_FINISHED = 4 + + +this.init = (task, methodResponse, cwmpResponse, deviceUpdates) -> + if methodResponse.faultcode? + task.fault = methodResponse + return STATUS_FAULT + if not task.subtask? task.subtask = {'name' : 'getParameterNames', 'parameterPath' : '', 'nextLevel' : false} if task.subtask.name == 'getParameterNames' - common.extend(ret, this.getParameterNames(task.subtask, methodResponse, cwmpResponse)) - if ret.parameterNames? # task finished + status = this.getParameterNames(task.subtask, methodResponse, cwmpResponse, deviceUpdates) + if status is STATUS_FINISHED parameterNames = [] - for p in ret.parameterNames + for p in deviceUpdates.parameterNames if not common.endsWith(p[0], '.') parameterNames.push(p[0]) task.subtask = {'name' : 'getParameterValues', 'parameterNames' : parameterNames} + else if status != STATUS_STARTED + throw Error('Unexpected subtask status') if task.subtask.name == 'getParameterValues' - common.extend(ret, this.getParameterValues(task.subtask, methodResponse, cwmpResponse)) + status = this.getParameterValues(task.subtask, methodResponse, cwmpResponse, deviceUpdates) + if status is STATUS_FINISHED + return STATUS_FINISHED + else if status is not STATUS_STARTED + throw Error('Unexpected subtask status') - ret + return STATUS_STARTED + + +this.getParameterNames = (task, methodResponse, cwmpResponse, deviceUpdates) -> + if methodResponse.faultcode? + task.fault = methodResponse + return STATUS_FAULT -this.getParameterNames = (task, methodResponse, cwmpResponse) -> if methodResponse.type is 'GetParameterNamesResponse' - return {'parameterNames' : methodResponse.parameterList} - else - cwmpResponse.methodRequest = { - type : 'GetParameterNames', - parameterPath : task.parameterPath, - nextLevel : if task.nextLevel? then task.nextLevel else false - } - return + deviceUpdates.parameterNames = methodResponse.parameterList + return STATUS_FINISHED + + cwmpResponse.methodRequest = { + type : 'GetParameterNames', + parameterPath : task.parameterPath, + nextLevel : if task.nextLevel? then task.nextLevel else false + } + return STATUS_STARTED + + +this.getParameterValues = (task, methodResponse, cwmpResponse, deviceUpdates) -> + if methodResponse.faultcode? + task.fault = methodResponse + return STATUS_FAULT -this.getParameterValues = (task, methodResponse, cwmpResponse) -> if not task.currentIndex? task.currentIndex = 0 else if methodResponse.parameterList? @@ -40,17 +67,25 @@ this.getParameterValues = (task, methodResponse, cwmpResponse) -> task.nextIndex = Math.min(task.currentIndex + BATCH_SIZE, task.parameterNames.length) names = task.parameterNames.slice(task.currentIndex, task.nextIndex) - if names.length > 0 - cwmpResponse.methodRequest = { - type : 'GetParameterValues', - parameterNames : names - } - if methodResponse.type is 'GetParameterValuesResponse' - return {'parameterValues' : methodResponse.parameterList} - return + deviceUpdates.parameterValues = methodResponse.parameterList + + if names.length == 0 + return STATUS_FINISHED + + cwmpResponse.methodRequest = { + type : 'GetParameterValues', + parameterNames : names + } + + return STATUS_STARTED + + +this.setParameterValues = (task, methodResponse, cwmpResponse, deviceUpdates) -> + if methodResponse.faultcode? + task.fault = methodResponse + return STATUS_FAULT -this.setParameterValues = (task, methodResponse, cwmpResponse) -> if not task.currentIndex? task.currentIndex = 0 else if methodResponse.type is 'SetParameterValuesResponse' @@ -60,29 +95,48 @@ this.setParameterValues = (task, methodResponse, cwmpResponse) -> task.nextIndex = Math.min(task.currentIndex + BATCH_SIZE, task.parameterValues.length) values = task.parameterValues.slice(task.currentIndex, task.nextIndex) - if values.length > 0 - cwmpResponse.methodRequest = { - type : 'SetParameterValues', - parameterList : values - } - if prevValues? - return {'parameterValues' : prevValues} - return + deviceUpdates.parameterValues = prevValues + + if values.length == 0 + return STATUS_FINISHED + + cwmpResponse.methodRequest = { + type : 'SetParameterValues', + parameterList : values + } + return STATUS_STARTED + + +this.reboot = (task, methodResponse, cwmpResponse, deviceUpdates) -> + if methodResponse.faultcode? + task.fault = methodResponse + return STATUS_FAULT -this.reboot = (task, methodResponse, cwmpResponse) -> if methodResponse.type isnt 'RebootResponse' cwmpResponse.methodRequest = { type : 'Reboot' } + return STATUS_STARTED + + return STATUS_FINISHED + + +this.download = (task, methodResponse, cwmpResponse, deviceUpdates) -> + if methodResponse.faultcode? + task.fault = methodResponse + return STATUS_FAULT -this.download = (task, methodResponse, cwmpResponse) -> if methodResponse.type isnt 'DownloadResponse' cwmpResponse.methodRequest = { type : 'Download', fileType : task.fileType, url : task.url } - -exports.task = (task, methodResponse, cwmpResponse) -> - return this[task.name](task, methodResponse, cwmpResponse) + return STATUS_STARTED + + return STATUS_FINISHED + + +exports.task = (task, methodResponse, cwmpResponse, deviceUpdates) -> + return this[task.name](task, methodResponse, cwmpResponse, deviceUpdates)