In case of fault response, store the fault and don't process any new task until fault is cleared.
This commit is contained in:
1 parent
d739adc32c
commit
60bda6921a
3 files changed
+48
-8
No files matched your search
+20
-7
@@ -73,7 +73,7 @@ updateDevice = (deviceId, actions, callback) ->
|
|||||||
return
|
return
|
||||||
|
|
||||||
if count == 0
|
if count == 0
|
||||||
util.log("#{currentRequest.deviceId}: new device detected")
|
util.log("#{currentRequest.deviceId}: New device detected")
|
||||||
db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {upsert: true, safe: true}, (err) ->
|
db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {upsert: true, safe: true}, (err) ->
|
||||||
if err?
|
if err?
|
||||||
callback(err) if callback?
|
callback(err) if callback?
|
||||||
@@ -107,7 +107,7 @@ runTask = (task, reqParams) ->
|
|||||||
# Task finished
|
# Task finished
|
||||||
db.memcached.del(String(task._id))
|
db.memcached.del(String(task._id))
|
||||||
db.tasksCollection.remove({'_id' : db.mongo.ObjectID(String(task._id))}, {safe: true}, (err, removed) ->
|
db.tasksCollection.remove({'_id' : db.mongo.ObjectID(String(task._id))}, {safe: true}, (err, removed) ->
|
||||||
util.log("#{currentRequest.deviceId}: completed task #{task.name}(#{task._id})")
|
util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})")
|
||||||
cookies = {task : null}
|
cookies = {task : null}
|
||||||
nextTask(cookies)
|
nextTask(cookies)
|
||||||
)
|
)
|
||||||
@@ -120,7 +120,7 @@ nextTask = (cookies) ->
|
|||||||
resParams = {}
|
resParams = {}
|
||||||
|
|
||||||
if not task
|
if not task
|
||||||
# no more taks, check presets discrepancy
|
# no more tasks, check presets discrepancy
|
||||||
db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) ->
|
db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) ->
|
||||||
presetsHash = results['presets_hash']
|
presetsHash = results['presets_hash']
|
||||||
devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"]
|
devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"]
|
||||||
@@ -146,8 +146,12 @@ nextTask = (cookies) ->
|
|||||||
writeResponse res
|
writeResponse res
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
|
else if task.fault?
|
||||||
util.log("#{currentRequest.deviceId}: started task #{task.name}(#{task._id})")
|
# last task was faulty. Do nothing until until task is deleted
|
||||||
|
res = tr069.response(currentRequest.sessionId, resParams, cookies)
|
||||||
|
writeResponse res
|
||||||
|
else
|
||||||
|
util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})")
|
||||||
runTask(task, reqParams)
|
runTask(task, reqParams)
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -175,7 +179,7 @@ else
|
|||||||
util.error(error.stack)
|
util.error(error.stack)
|
||||||
# dump request/response logs and stack trace
|
# dump request/response logs and stack trace
|
||||||
db.memcached.get("debug-#{currentRequest.clientIp}", (err, l) ->
|
db.memcached.get("debug-#{currentRequest.clientIp}", (err, l) ->
|
||||||
util.log("Unexpected error occured. Writing log to debug/#{currentRequest.clientIp}.log.")
|
util.log("#{currentRequest.deviceId}: Unexpected error occured. Writing log to debug/#{currentRequest.clientIp}.log.")
|
||||||
util.error(err) if err
|
util.error(err) if err
|
||||||
fs = require 'fs'
|
fs = require 'fs'
|
||||||
fs.writeFileSync("debug/#{currentRequest.clientIp}.log", l + "\n\n" + error.stack)
|
fs.writeFileSync("debug/#{currentRequest.clientIp}.log", l + "\n\n" + error.stack)
|
||||||
@@ -231,7 +235,7 @@ else
|
|||||||
cookies.ID = currentRequest.sessionId = reqParams.sessionId
|
cookies.ID = currentRequest.sessionId = reqParams.sessionId
|
||||||
cookies.DeviceId = currentRequest.deviceId = common.getDeviceId(reqParams.deviceId)
|
cookies.DeviceId = currentRequest.deviceId = common.getDeviceId(reqParams.deviceId)
|
||||||
if config.DEBUG
|
if config.DEBUG
|
||||||
util.log("#{currentRequest.deviceId}: inform (#{reqParams.eventCodes}); retry count #{reqParams.retryCount}")
|
util.log("#{currentRequest.deviceId}: Inform (#{reqParams.eventCodes}); retry count #{reqParams.retryCount}")
|
||||||
|
|
||||||
updateDevice(currentRequest.deviceId, {'inform' : true, 'parameterValues' : reqParams.informParameterValues}, (err) ->
|
updateDevice(currentRequest.deviceId, {'inform' : true, 'parameterValues' : reqParams.informParameterValues}, (err) ->
|
||||||
res = tr069.response(currentRequest.sessionId, resParams, cookies)
|
res = tr069.response(currentRequest.sessionId, resParams, cookies)
|
||||||
@@ -247,6 +251,15 @@ else
|
|||||||
nextTask(cookies)
|
nextTask(cookies)
|
||||||
else
|
else
|
||||||
db.getTask(taskId, (task) ->
|
db.getTask(taskId, (task) ->
|
||||||
|
if reqParams.fault?
|
||||||
|
util.log("#{currentRequest.deviceId}: Fault response for task #{task._id}")
|
||||||
|
task.fault = reqParams.fault
|
||||||
|
db.saveTask(task, (err) ->
|
||||||
|
# Faulty task. No more work to do until task is deleted.
|
||||||
|
res = tr069.response(currentRequest.sessionId, resParams, cookies)
|
||||||
|
writeResponse res
|
||||||
|
)
|
||||||
|
else
|
||||||
runTask(task, reqParams)
|
runTask(task, reqParams)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -30,6 +30,7 @@ db.open( (err, db) ->
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
getTask = (taskId, callback) ->
|
getTask = (taskId, callback) ->
|
||||||
memcached.get(taskId, (err, task) ->
|
memcached.get(taskId, (err, task) ->
|
||||||
if not task?
|
if not task?
|
||||||
@@ -55,7 +56,15 @@ updateTask = (task, callback) ->
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
saveTask = (task, callback) ->
|
||||||
|
task._id = mongo.ObjectID(String(task._id))
|
||||||
|
tasksCollection.save(task, (err) ->
|
||||||
|
callback(err)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
exports.mongo = mongo
|
exports.mongo = mongo
|
||||||
exports.memcached = memcached
|
exports.memcached = memcached
|
||||||
exports.getTask = getTask
|
exports.getTask = getTask
|
||||||
exports.updateTask = updateTask
|
exports.updateTask = updateTask
|
||||||
|
exports.saveTask = saveTask
|
||||||
+19
-1
@@ -116,6 +116,23 @@ setParameterValuesResponse = (xml) ->
|
|||||||
return JSON.parse(e.text())
|
return JSON.parse(e.text())
|
||||||
return undefined
|
return undefined
|
||||||
|
|
||||||
|
fault = (xml) ->
|
||||||
|
f = xml.get('//soap-env:Envelope/soap-env:Body/soap-env:Fault', NAMESPACES)
|
||||||
|
if f
|
||||||
|
traverse = (el) ->
|
||||||
|
children = el.childNodes()
|
||||||
|
obj = {}
|
||||||
|
for n in el.childNodes()
|
||||||
|
if n.type() == 'element'
|
||||||
|
obj[n.name()] = traverse(n)
|
||||||
|
if Object.keys(obj).length == 0
|
||||||
|
return el.text()
|
||||||
|
else
|
||||||
|
return obj
|
||||||
|
|
||||||
|
return traverse(f)
|
||||||
|
return undefined
|
||||||
|
|
||||||
|
|
||||||
exports.request = (request) ->
|
exports.request = (request) ->
|
||||||
req = {cookies: {}}
|
req = {cookies: {}}
|
||||||
@@ -140,6 +157,7 @@ exports.request = (request) ->
|
|||||||
req.getParameterNamesResponse = getParameterNamesResponse xml
|
req.getParameterNamesResponse = getParameterNamesResponse xml
|
||||||
req.getParameterValuesResponse = getParameterValuesResponse xml
|
req.getParameterValuesResponse = getParameterValuesResponse xml
|
||||||
req.setParameterValuesResponse = setParameterValuesResponse xml
|
req.setParameterValuesResponse = setParameterValuesResponse xml
|
||||||
|
req.fault = fault xml
|
||||||
return req
|
return req
|
||||||
|
|
||||||
|
|
||||||
@@ -157,7 +175,7 @@ exports.response = (sessionId, params, cookies = null) ->
|
|||||||
#console.log '>>> EMPTY RESPONSE'
|
#console.log '>>> EMPTY RESPONSE'
|
||||||
# send empty response
|
# send empty response
|
||||||
headers['Content-Length'] = 0
|
headers['Content-Length'] = 0
|
||||||
return {code: 204, headers: headers, data: null}
|
return {code: 204, headers: headers, data: ''}
|
||||||
|
|
||||||
xml = libxmljs.Document()
|
xml = libxmljs.Document()
|
||||||
env = xml.node('soap-env:Envelope')
|
env = xml.node('soap-env:Envelope')
|
||||||
|
|||||||
Reference in new issue
Block a user