rodrigo/genieacs
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Saving task fault and session data separately to avoid persistant session data when a fault occures.

Browse Source
This commit is contained in:
Zaid Abdulla committed 2013-11-02 15:51:53 +03:00
1 parent fd378d138a
commit 5a2648c4d0
3 files changed
+115 -130
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

acs.coffee
+28 -25
View File
@@ -152,40 +152,43 @@ runTask = (currentRequest, task, methodResponse) ->
# Server under load. Hold new sessions temporarily. # Server under load. Hold new sessions temporarily.
holdUntil = Math.max(Date.now() + timeDiff * 2000, holdUntil) holdUntil = Math.max(Date.now() + timeDiff * 2000, holdUntil)
switch status save = status & tasks.STATUS_SAVE
when tasks.STATUS_FINISHED switch status & ~tasks.STATUS_SAVE
db.memcached.del(String(task._id)) when tasks.STATUS_OK
db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) -> f = () ->
throw new Error(err) if err? db.memcached.set(String(task._id), task, config.CACHE_DURATION, (err, res) ->
throw err if err
res = tr069.response(task._id, cwmpResponse)
writeResponse(currentRequest, res)
)
if save
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.deviceId}: Completed task #{task.name}(#{task._id})") util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})")
db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) ->
throw err if err
db.memcached.del(String(task._id), (err, res) ->
nextTask(currentRequest) nextTask(currentRequest)
) )
)
when tasks.STATUS_FAULT when tasks.STATUS_FAULT
util.log("#{currentRequest.deviceId}: Fault response for task #{task._id}") util.log("#{currentRequest.deviceId}: Fault response for task #{task._id}")
db.saveTask(task, (err) -> taskUpdate = {fault : task.fault}
throw new Error(err) if err? if save
taskUpdate.session = task.session
db.tasksCollection.update({_id : mongodb.ObjectID(String(task._id))}, {$set : taskUpdate}, (err) ->
throw err if err
# Faulty task. No more work to do until task is deleted. # Faulty task. No more work to do until task is deleted.
res = tr069.response(null, cwmpResponse) res = tr069.response(null, cwmpResponse)
writeResponse(currentRequest, res) writeResponse(currentRequest, res)
) )
when tasks.STATUS_PENDING
db.saveTask(task, (err) ->
throw new Error(err) if err?
# task expects CPE confirmation later
nextTask(currentRequest)
)
when tasks.STATUS_STARTED
db.updateTask(task, (err) ->
throw new Error(err) if err?
res = tr069.response(task._id, cwmpResponse)
writeResponse(currentRequest, res)
)
when tasks.STATUS_SAVE
db.saveTask(task, (err) ->
throw new Error(err) if err?
res = tr069.response(task._id, cwmpResponse)
writeResponse(currentRequest, res)
)
else else
throw Error('Unknown task status') throw Error('Unknown task status')
) )
db.coffee
-27
View File
@@ -57,31 +57,6 @@ getTask = (taskId, callback) ->
) )
updateTask = (task, callback) ->
id = String(task._id)
memcached.set(id, task, config.CACHE_DURATION, (err, res) ->
if res
callback()
else
task._id = mongodb.ObjectID(id)
tasksCollection.save(task, (err) ->
callback(err)
)
)
saveTask = (task, callback) ->
# task ID can either be a string or a MongoDB ID
id = String(task._id)
task._id = mongodb.ObjectID(id)
tasksCollection.save(task, (mongoErr) ->
memcached.del(id, (memcachedErr, res) ->
callback(mongoErr or memcachedErr)
)
)
getPresets = (callback) -> getPresets = (callback) ->
memcached.get(['presets', 'objects'], (err, res) -> memcached.get(['presets', 'objects'], (err, res) ->
presets = res.presets presets = res.presets
@@ -112,6 +87,4 @@ getPresets = (callback) ->
exports.memcached = memcached exports.memcached = memcached
exports.getTask = getTask exports.getTask = getTask
exports.updateTask = updateTask
exports.saveTask = saveTask
exports.getPresets = getPresets exports.getPresets = getPresets
tasks.coffee
+87 -78
View File
@@ -5,12 +5,10 @@ mongodb = require 'mongodb'
customCommands = require './custom-commands' customCommands = require './custom-commands'
BATCH_SIZE = 16 BATCH_SIZE = 16
exports.STATUS_QUEUED = STATUS_QUEUED = 0 exports.STATUS_OK = STATUS_OK = 1
exports.STATUS_STARTED = STATUS_STARTED = 1 exports.STATUS_FAULT = STATUS_FAULT = 2
exports.STATUS_SAVE = STATUS_SAVE = 2 exports.STATUS_COMPLETED = STATUS_COMPLETED = 4
exports.STATUS_PENDING = STATUS_PENDING = 3 exports.STATUS_SAVE = STATUS_SAVE = 128 # not mutually exclusive with the rest
exports.STATUS_FAULT = STATUS_FAULT = 4
exports.STATUS_FINISHED = STATUS_FINISHED = 5
initCustomCommands = (deviceId, callback) -> initCustomCommands = (deviceId, callback) ->
@@ -27,37 +25,40 @@ initCustomCommands = (deviceId, callback) ->
this.init = (task, methodResponse, callback) -> this.init = (task, methodResponse, callback) ->
if not task.subtask? task.session = {} if not task.session?
task.subtask = {device : task.device, name : 'getParameterNames', parameterPath : '', nextLevel : false}
if task.subtask.name == 'getParameterNames' if not task.session.subtask?
task.session.subtask = {device : task.device, name : 'getParameterNames', parameterPath : '', nextLevel : false}
if task.session.subtask.name is 'getParameterNames'
if methodResponse.faultcode? if methodResponse.faultcode?
task.fault = methodResponse task.fault = methodResponse
callback(null, STATUS_FAULT) callback(null, STATUS_FAULT)
return return
this.getParameterNames(task.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) => this.getParameterNames(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
if status is STATUS_FINISHED if status & STATUS_COMPLETED
parameterNames = [] parameterNames = []
for p in deviceUpdates.parameterNames for p in deviceUpdates.parameterNames
if not common.endsWith(p[0], '.') if not common.endsWith(p[0], '.')
parameterNames.push(p[0]) parameterNames.push(p[0])
task.subtask = {name : 'getParameterValues', parameterNames : parameterNames} task.session.subtask = {name : 'getParameterValues', parameterNames : parameterNames}
this.getParameterValues(task.subtask, {}, (err, status, cwmpResponse) -> this.getParameterValues(task.session.subtask, {}, (err, status, cwmpResponse) ->
# ignore deviceUpdates returned by firt call to getParameterValues # ignore deviceUpdates returned by firt call to getParameterValues
callback(err, status, cwmpResponse, deviceUpdates) callback(err, status, cwmpResponse, deviceUpdates)
) )
else if status is STATUS_STARTED else if status & STATUS_OK
callback(err, STATUS_STARTED, cwmpResponse, deviceUpdates) callback(err, STATUS_OK, cwmpResponse, deviceUpdates)
else else
throw Error('Unexpected subtask status') throw Error('Unexpected subtask status')
) )
else if task.subtask.name == 'getParameterValues' else if task.session.subtask.name is 'getParameterValues'
if methodResponse.faultcode? if methodResponse.faultcode?
# Ignore GetParameterValues errors. A workaround for the crappy Seewon devices. # Ignore GetParameterValues errors. A workaround for the crappy Seewon devices.
methodResponse = {parameterList : {}} methodResponse = {parameterList : {}}
this.getParameterValues(task.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) ->
if status is STATUS_FINISHED this.getParameterValues(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) ->
if status & STATUS_COMPLETED
initCustomCommands(task.device, (updates) -> initCustomCommands(task.device, (updates) ->
common.extend(deviceUpdates, updates) common.extend(deviceUpdates, updates)
callback(err, status, cwmpResponse, deviceUpdates) callback(err, status, cwmpResponse, deviceUpdates)
@@ -70,36 +71,39 @@ this.init = (task, methodResponse, callback) ->
this.refreshObject = (task, methodResponse, callback) -> this.refreshObject = (task, methodResponse, callback) ->
if not task.subtask? task.session = {} if not task.session?
task.subtask = {device : task.device, name : 'getParameterNames', parameterPath : "#{task.objectName}.", nextLevel : false}
if not task.session.subtask?
task.session.subtask = {device : task.device, name : 'getParameterNames', parameterPath : "#{task.objectName}.", nextLevel : false}
if methodResponse.faultcode? if methodResponse.faultcode?
task.fault = methodResponse task.fault = methodResponse
callback(null, STATUS_FAULT) callback(null, STATUS_FAULT)
return return
if task.subtask.name == 'getParameterNames' if task.session.subtask.name is 'getParameterNames'
this.getParameterNames(task.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) => this.getParameterNames(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
if status is STATUS_FINISHED if status & STATUS_COMPLETED
parameterNames = [] parameterNames = []
for p in deviceUpdates.parameterNames for p in deviceUpdates.parameterNames
if not common.endsWith(p[0], '.') if not common.endsWith(p[0], '.')
parameterNames.push(p[0]) parameterNames.push(p[0])
task.subtask = {name : 'getParameterValues', parameterNames : parameterNames} task.session.subtask = {name : 'getParameterValues', parameterNames : parameterNames}
this.getParameterValues(task.subtask, {}, (err, status, cwmpResponse) -> this.getParameterValues(task.session.subtask, {}, (err, status, cwmpResponse) ->
# ignore deviceUpdates returned by firt call to getParameterValues # ignore deviceUpdates returned by firt call to getParameterValues
callback(err, status, cwmpResponse, deviceUpdates) callback(err, status, cwmpResponse, deviceUpdates)
) )
else if status is STATUS_STARTED else if status & STATUS_OK
callback(err, STATUS_STARTED, cwmpResponse, deviceUpdates) callback(err, STATUS_OK, cwmpResponse, deviceUpdates)
else else
throw Error('Unexpected subtask status') throw Error('Unexpected subtask status')
) )
else if task.subtask.name == 'getParameterValues' else if task.session.subtask.name is 'getParameterValues'
if methodResponse.faultcode? if methodResponse.faultcode?
# Ignore GetParameterValues errors. A workaround for the crappy Seewon devices. # Ignore GetParameterValues errors. A workaround for the crappy Seewon devices.
methodResponse = {parameterList : {}} methodResponse = {parameterList : {}}
this.getParameterValues(task.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) ->
this.getParameterValues(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) ->
callback(err, status, cwmpResponse, deviceUpdates) callback(err, status, cwmpResponse, deviceUpdates)
) )
else else
@@ -168,7 +172,7 @@ this.getParameterNames = (task, methodResponse, callback) ->
# some devices don't return the root object as described in the standard. add manually to update timestamp # some devices don't return the root object as described in the standard. add manually to update timestamp
deviceUpdates.parameterNames.push([task.parameterPath]) if !!task.parameterPath deviceUpdates.parameterNames.push([task.parameterPath]) if !!task.parameterPath
callback(null, STATUS_FINISHED, null, deviceUpdates) callback(null, STATUS_COMPLETED, null, deviceUpdates)
) )
else else
methodRequest = { methodRequest = {
@@ -176,61 +180,67 @@ this.getParameterNames = (task, methodResponse, callback) ->
parameterPath : task.parameterPath, parameterPath : task.parameterPath,
nextLevel : if task.nextLevel? then task.nextLevel else false nextLevel : if task.nextLevel? then task.nextLevel else false
} }
callback(null, STATUS_STARTED, {methodRequest : methodRequest}) callback(null, STATUS_OK, {methodRequest : methodRequest})
this.getParameterValues = (task, methodResponse, callback) -> this.getParameterValues = (task, methodResponse, callback) ->
task.session = {} if not task.session?
if methodResponse.faultcode? if methodResponse.faultcode?
task.fault = methodResponse task.fault = methodResponse
callback(null, STATUS_FAULT) callback(null, STATUS_FAULT)
return return
if not task.currentIndex? if not task.session.currentIndex?
task.currentIndex = 0 task.session.currentIndex = 0
else if methodResponse.parameterList? else if methodResponse.parameterList?
task.currentIndex = task.nextIndex task.session.currentIndex = task.session.nextIndex
task.nextIndex = Math.min(task.currentIndex + BATCH_SIZE, task.parameterNames.length) task.session.nextIndex = Math.min(task.session.currentIndex + BATCH_SIZE, task.parameterNames.length)
names = task.parameterNames.slice(task.currentIndex, task.nextIndex) names = task.parameterNames.slice(task.session.currentIndex, task.session.nextIndex)
if methodResponse.type is 'GetParameterValuesResponse' if methodResponse.type is 'GetParameterValuesResponse'
deviceUpdates = {parameterValues : methodResponse.parameterList} deviceUpdates = {parameterValues : methodResponse.parameterList}
if names.length == 0 if names.length == 0
callback(null, STATUS_FINISHED, null, deviceUpdates) callback(null, STATUS_COMPLETED, null, deviceUpdates)
else else
methodRequest = { methodRequest = {
type : 'GetParameterValues', type : 'GetParameterValues',
parameterNames : names parameterNames : names
} }
callback(null, STATUS_STARTED, {methodRequest : methodRequest}, deviceUpdates) callback(null, STATUS_OK, {methodRequest : methodRequest}, deviceUpdates)
this.setParameterValues = (task, methodResponse, callback) -> this.setParameterValues = (task, methodResponse, callback) ->
task.session = {} if not task.session?
if methodResponse.faultcode? if methodResponse.faultcode?
task.fault = methodResponse task.fault = methodResponse
callback(null, STATUS_FAULT) callback(null, STATUS_FAULT)
return return
if not task.currentIndex? if not task.session.currentIndex?
task.currentIndex = 0 task.session.currentIndex = 0
else if methodResponse.type is 'SetParameterValuesResponse' else if methodResponse.type is 'SetParameterValuesResponse'
prevValues = task.parameterValues.slice(task.currentIndex, task.nextIndex) prevValues = task.parameterValues.slice(task.session.currentIndex, task.session.nextIndex)
task.currentIndex = task.nextIndex task.session.currentIndex = task.session.nextIndex
task.nextIndex = Math.min(task.currentIndex + BATCH_SIZE, task.parameterValues.length) task.session.nextIndex = Math.min(task.session.currentIndex + BATCH_SIZE, task.parameterValues.length)
values = task.parameterValues.slice(task.currentIndex, task.nextIndex) values = task.parameterValues.slice(task.session.currentIndex, task.session.nextIndex)
if prevValues? if prevValues?
deviceUpdates = {parameterValues : prevValues} deviceUpdates = {parameterValues : prevValues}
if values.length == 0 if values.length == 0
callback(null, STATUS_FINISHED, null, deviceUpdates) callback(null, STATUS_COMPLETED, null, deviceUpdates)
else else
callback(null, STATUS_STARTED, {methodRequest : {type : 'SetParameterValues', parameterList : values}}, deviceUpdates) callback(null, STATUS_OK, {methodRequest : {type : 'SetParameterValues', parameterList : values}}, deviceUpdates)
this.addObject = (task, methodResponse, callback) -> this.addObject = (task, methodResponse, callback) ->
task.session = {} if not task.session?
if methodResponse.faultcode? if methodResponse.faultcode?
task.fault = methodResponse task.fault = methodResponse
callback(null, STATUS_FAULT) callback(null, STATUS_FAULT)
@@ -238,40 +248,38 @@ this.addObject = (task, methodResponse, callback) ->
allDeviceUpdates = {} allDeviceUpdates = {}
if not task.instanceNumber? if not task.session.instanceNumber?
if methodResponse.type is 'AddObjectResponse' if methodResponse.type is 'AddObjectResponse'
task.instanceNumber = methodResponse.instanceNumber status_save = STATUS_SAVE
task.subtask = {device : task.device, name : 'getParameterNames', parameterPath : "#{task.objectName}.#{task.instanceNumber}.", nextLevel : false} task.session.instanceNumber = methodResponse.instanceNumber
task.appliedParameterValues = [] task.session.subtask = {device : task.device, name : 'getParameterNames', parameterPath : "#{task.objectName}.#{task.session.instanceNumber}.", nextLevel : false}
task.parameterNames = [] task.session.appliedParameterValues = []
allDeviceUpdates.instanceName = [["#{task.objectName}.#{task.instanceNumber}", task.instanceName]] if task.instanceName? task.session.parameterNames = []
allDeviceUpdates.instanceName = [["#{task.objectName}.#{task.session.instanceNumber}", task.instanceName]] if task.instanceName?
else else
callback(null, STATUS_STARTED, {methodRequest : {type : 'AddObject', objectName : "#{task.objectName}."}}) callback(null, STATUS_OK, {methodRequest : {type : 'AddObject', objectName : "#{task.objectName}."}})
return return
subtask = () => subtask = () =>
switch task.subtask.name switch task.session.subtask.name
when 'getParameterNames' when 'getParameterNames'
this.getParameterNames(task.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) => this.getParameterNames(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
common.extend(allDeviceUpdates, deviceUpdates) common.extend(allDeviceUpdates, deviceUpdates)
if deviceUpdates and deviceUpdates.parameterNames if deviceUpdates and deviceUpdates.parameterNames
for p in deviceUpdates.parameterNames for p in deviceUpdates.parameterNames
task.parameterNames.push(p[0]) if not common.endsWith(p[0], '.') task.session.parameterNames.push(p[0]) if not common.endsWith(p[0], '.')
if status is STATUS_FINISHED if status & STATUS_COMPLETED
task.subtask = {name : 'getParameterValues', parameterNames : task.parameterNames} task.session.subtask = {name : 'getParameterValues', parameterNames : task.session.parameterNames}
subtask() subtask()
else if status is STATUS_STARTED else if status & STATUS_OK
if allDeviceUpdates.instanceName? # Use STATUS_SAVE to avoid adding duplicate object in case of error
# Use STATUSS_SAVE to avoid adding duplicate object in case of error callback(err, STATUS_OK | status_save, cwmpResponse, allDeviceUpdates)
callback(err, STATUS_SAVE, cwmpResponse, allDeviceUpdates)
else
callback(err, STATUS_STARTED, cwmpResponse, allDeviceUpdates)
else else
throw Error('Unexpected subtask status') throw Error('Unexpected subtask status')
) )
when 'getParameterValues' when 'getParameterValues'
this.getParameterValues(task.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) => this.getParameterValues(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
common.extend(allDeviceUpdates, deviceUpdates) common.extend(allDeviceUpdates, deviceUpdates)
# if values are given, compare with default values # if values are given, compare with default values
if task.parameterValues? if task.parameterValues?
@@ -282,19 +290,20 @@ this.addObject = (task, methodResponse, callback) ->
t = if p2[2] then p2[2] else p1[2] t = if p2[2] then p2[2] else p1[2]
v = common.matchType(p1[1], p2[1]) v = common.matchType(p1[1], p2[1])
# TODO only include if writable # TODO only include if writable
task.appliedParameterValues.push([p1[0], v, t]) task.session.appliedParameterValues.push([p1[0], v, t])
if methodResponse.faultcode? if methodResponse.faultcode?
# Ignore GetParameterValues errors. A workaround for the crappy Seewon devices. # Ignore GetParameterValues errors. A workaround for the crappy Seewon devices.
methodResponse = {parameterList : {}} methodResponse = {parameterList : {}}
if status is STATUS_FINISHED and task.appliedParameterValues.length > 0
task.subtask = {name : 'setParameterValues', parameterValues : task.appliedParameterValues} if status & STATUS_COMPLETED and task.session.appliedParameterValues.length > 0
task.session.subtask = {name : 'setParameterValues', parameterValues : task.session.appliedParameterValues}
subtask() subtask()
else else
callback(err, status, cwmpResponse, allDeviceUpdates) callback(err, status, cwmpResponse, allDeviceUpdates)
) )
when 'setParameterValues' when 'setParameterValues'
this.setParameterValues(task.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) => this.setParameterValues(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
common.extend(allDeviceUpdates, deviceUpdates) common.extend(allDeviceUpdates, deviceUpdates)
callback(err, status, cwmpResponse, allDeviceUpdates) callback(err, status, cwmpResponse, allDeviceUpdates)
) )
@@ -308,13 +317,13 @@ this.deleteObject = (task, methodResponse, callback) ->
return return
if methodResponse.type is 'DeleteObjectResponse' if methodResponse.type is 'DeleteObjectResponse'
callback(null, STATUS_FINISHED, null, {deletedObjects : [task.objectName]}) callback(null, STATUS_COMPLETED, null, {deletedObjects : [task.objectName]})
else else
methodRequest = { methodRequest = {
type : 'DeleteObject', type : 'DeleteObject',
objectName : "#{task.objectName}." objectName : "#{task.objectName}."
} }
callback(null, STATUS_STARTED, {methodRequest : methodRequest}) callback(null, STATUS_OK, {methodRequest : methodRequest})
this.reboot = (task, methodResponse, callback) -> this.reboot = (task, methodResponse, callback) ->
@@ -324,9 +333,9 @@ this.reboot = (task, methodResponse, callback) ->
return return
if methodResponse.type isnt 'RebootResponse' if methodResponse.type isnt 'RebootResponse'
callback(null, STATUS_STARTED, {methodRequest : {type : 'Reboot'}}) callback(null, STATUS_OK, {methodRequest : {type : 'Reboot'}})
else else
callback(null, STATUS_FINISHED) callback(null, STATUS_COMPLETED)
this.factoryReset = (task, methodResponse, callback) -> this.factoryReset = (task, methodResponse, callback) ->
@@ -336,9 +345,9 @@ this.factoryReset = (task, methodResponse, callback) ->
return return
if methodResponse.type isnt 'FactoryResetResponse' if methodResponse.type isnt 'FactoryResetResponse'
callback(null, STATUS_STARTED, {methodRequest : {type : 'FactoryReset'}}) callback(null, STATUS_OK, {methodRequest : {type : 'FactoryReset'}})
else else
callback(null, STATUS_FINISHED) callback(null, STATUS_COMPLETED)
this.download = (task, methodResponse, callback) -> this.download = (task, methodResponse, callback) ->
@@ -364,10 +373,10 @@ this.download = (task, methodResponse, callback) ->
successUrl : task.successUrl, successUrl : task.successUrl,
failureUrl : task.failureUrl failureUrl : task.failureUrl
} }
callback(null, STATUS_STARTED, {methodRequest : methodRequest}) callback(null, STATUS_OK, {methodRequest : methodRequest})
) )
else else
callback(null, STATUS_FINISHED) callback(null, STATUS_COMPLETED)
this.customCommand = (task, methodResponse, callback) -> this.customCommand = (task, methodResponse, callback) ->
@@ -378,7 +387,7 @@ this.customCommand = (task, methodResponse, callback) ->
callback(null, STATUS_FAULT) callback(null, STATUS_FAULT)
else else
commandName = task.command.split(' ', 2)[0] commandName = task.command.split(' ', 2)[0]
callback(null, STATUS_FINISHED, null, {customCommands : [[commandName, value]]}) callback(null, STATUS_COMPLETED, null, {customCommands : [[commandName, value]]})
) )