Generate faults in precondition testing stage

Also, queue applicable provisions to any existing fault in order to
be processed later.
This commit is contained in:
Zaid Abdulla committed 2016-12-21 19:44:56 -08:00
1 parent c4880b636b
commit 63407be19b
3 files changed
+276 -262

No files matched your search

+169 -72
View File
@@ -83,6 +83,41 @@ writeResponse = (currentRequest, res) ->
currentRequest.httpResponse.end(res.data) currentRequest.httpResponse.end(res.data)
recordFault = (sessionData, fault, provisions, channels) ->
faults = sessionData.faults
for channel of channels
provs = sessionData.faults[channel]?.provisions or []
faults[channel] = Object.assign({provisions: provs}, fault)
if sessionData.retries[channel]?
++ sessionData.retries[channel]
else
sessionData.retries[channel] = 0
if Object.keys(channels).length != 1
faults[channel].retryNow = true
if channels[channel] == 0
faults[channel].precondition = true
sessionData.faultsTouched ?= {}
sessionData.faultsTouched[channel] = true
util.log("#{sessionData.deviceId}: Fault occurred in channel #{channel} (retries: #{sessionData.retries[channel]})")
for provision, i in provisions
for channel of channels
if (channels[channel] >> i) & 1
faults[channel].provisions.push(provision)
for channel of channels
provs = faults[channel].provisions
faults[channel].provisions = []
appendProvisions(faults[channel].provisions, provs)
session.clearProvisions(sessionData)
return
inform = (currentRequest, cwmpRequest) -> inform = (currentRequest, cwmpRequest) ->
if config.get('LOG_INFORMS', currentRequest.sessionData.deviceId) if config.get('LOG_INFORMS', currentRequest.sessionData.deviceId)
util.log("#{currentRequest.sessionData.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") util.log("#{currentRequest.sessionData.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}")
@@ -105,15 +140,14 @@ inform = (currentRequest, cwmpRequest) ->
transferComplete = (currentRequest, cwmpRequest) -> transferComplete = (currentRequest, cwmpRequest) ->
session.transferComplete(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse, faults) -> session.transferComplete(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse, fault, operation) ->
throw err if err throw err if err
for channel, fault of faults if fault
if not fault.retries? and Object.keys(faults).length == 1 for k, v of operation.retries
fault.retries = 0 currentRequest.sessionData.retries[k] = v
currentRequest.sessionData.faults[channel] = fault recordFault(currentRequest.sessionData, fault, operation.provisions, operation.channels)
util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})")
res = soap.response({ res = soap.response({
id : cwmpRequest.id, id : cwmpRequest.id,
@@ -146,12 +180,37 @@ testSchedule = (schedule, timestamp) ->
return [false, first - timestamp] return [false, first - timestamp]
# Append providions and remove duplicates
appendProvisions = (original, toAppend) ->
modified = false
stringified = new WeakMap()
for p, i in original
stringified.set(p, JSON.stringify(p))
for p, i in toAppend by -1
s = JSON.stringify(p)
for pp, j in original by -1
ss = stringified.get(pp)
if s == ss
if (not p) or j >= original.length - (toAppend.length - i)
p = null
else
original.splice(j, 1)
modified = true
if p
original.splice(original.length - (toAppend.length - i) + 1, 0, p)
stringified.set(p, s)
modified = true
return modified
applyPresets = (currentRequest) -> applyPresets = (currentRequest) ->
cache.getPresets((err, presetsHash, presets) -> cache.getPresets((err, presetsHash, presets) ->
throw err if err throw err if err
# Filter presets based on existing faults # Filter presets based on existing faults
blackList = null blackList = {}
whiteList = null whiteList = null
whiteListProvisions = null whiteListProvisions = null
@@ -159,18 +218,17 @@ applyPresets = (currentRequest) ->
if currentRequest.sessionData.faults? if currentRequest.sessionData.faults?
for channel, fault of currentRequest.sessionData.faults for channel, fault of currentRequest.sessionData.faults
if fault.retries? if fault.retryNow
retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, fault.retries)) * 1000
else
retryTimestamp = 0 retryTimestamp = 0
else
retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, currentRequest.sessionData.retries[channel])) * 1000
if retryTimestamp <= currentRequest.sessionData.timestamp if retryTimestamp <= currentRequest.sessionData.timestamp
whiteList = channel whiteList = channel
whiteListProvisions = fault.provisions whiteListProvisions = fault.provisions
break break
blackList ?= [] blackList[channel] = if fault.precondition then 1 else 2
blackList.push(channel)
currentRequest.sessionData.presetsHash = presetsHash currentRequest.sessionData.presetsHash = presetsHash
@@ -183,23 +241,24 @@ applyPresets = (currentRequest) ->
parameters = {} parameters = {}
filteredPresets = [] filteredPresets = []
for p in presets for preset in presets
if whiteList? if whiteList?
continue if p.channel != whiteList continue if preset.channel != whiteList
else if blackList and p.channel in blackList else if blackList[channel] == 1
continue continue
eventsMatch = true eventsMatch = true
for k, v of p.events for k, v of preset.events
if (!v) != (!deviceEvents[k.replace(' ', '_')]) if (!v) != (!deviceEvents[k.replace(' ', '_')])
eventsMatch = false eventsMatch = false
break break
continue if (not eventsMatch) or (p.schedule and not continue if (not eventsMatch) or (preset.schedule and not
(p.schedule.schedule and testSchedule(p.schedule, currentRequest.sessionData.timestamp)[0])) (preset.schedule.schedule and testSchedule(preset.schedule, currentRequest.sessionData.timestamp)[0]))
filteredPresets.push(p) filteredPresets.push(preset)
for k of p.precondition for k of preset.precondition
currentRequest.sessionData.channels[preset.channel] = 0
p = k.split(/([^a-zA-Z0-9\-\_\.].*)/, 1)[0] p = k.split(/([^a-zA-Z0-9\-\_\.].*)/, 1)[0]
parameters[p] = common.parsePath(p) parameters[p] = common.parsePath(p)
@@ -207,15 +266,19 @@ applyPresets = (currentRequest) ->
for k, v of parameters for k, v of parameters
declarations.push([v, 1, {value: 1}]) declarations.push([v, 1, {value: 1}])
session.rpcRequest(currentRequest.sessionData, declarations, (err, faults, id, rpcRequest) -> session.rpcRequest(currentRequest.sessionData, declarations, (err, fault, id, rpcRequest) ->
throw err if err throw err if err
if faults? if fault
throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(faults)) recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels)
session.clearProvisions(currentRequest.sessionData)
return applyPresets(currentRequest)
if rpcRequest? if rpcRequest?
return sendRpcRequest(currentRequest, id, rpcRequest) return sendRpcRequest(currentRequest, id, rpcRequest)
session.clearProvisions(currentRequest.sessionData)
for k, v of parameters for k, v of parameters
unpacked = device.unpack(currentRequest.sessionData.deviceData, v) unpacked = device.unpack(currentRequest.sessionData.deviceData, v)
if unpacked[0] and (vv = currentRequest.sessionData.deviceData.attributes.get(unpacked[0]).value?[1])? if unpacked[0] and (vv = currentRequest.sessionData.deviceData.attributes.get(unpacked[0]).value?[1])?
@@ -226,24 +289,37 @@ applyPresets = (currentRequest) ->
if whiteList? if whiteList?
session.addProvisions(currentRequest.sessionData, whiteList, whiteListProvisions) session.addProvisions(currentRequest.sessionData, whiteList, whiteListProvisions)
appendProvisionsToFaults = {}
for p in filteredPresets for p in filteredPresets
if query.testFilter(parameters, p.precondition) if query.testFilter(parameters, p.precondition)
session.addProvisions(currentRequest.sessionData, p.channel, p.provisions) if blackList[p.channel] == 2
appendProvisionsToFaults[p.channel] =
(appendProvisionsToFaults[p.channel] or []).concat(p.provisions)
else
session.addProvisions(currentRequest.sessionData, p.channel, p.provisions)
session.rpcRequest(currentRequest.sessionData, null, (err, faults, id, rpcRequest, doneProvisions) -> for channel, provisions of appendProvisionsToFaults
if appendProvisions(currentRequest.sessionData.faults[channel].provisions, provisions)
currentRequest.sessionData.faultsTouched ?= {}
currentRequest.sessionData.faultsTouched[channel] = true
session.rpcRequest(currentRequest.sessionData, null, (err, fault, id, rpcRequest) ->
throw err if err throw err if err
if faults? if fault
for channel, fault of faults recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels)
if not fault.retries? and Object.keys(faults).length == 1 session.clearProvisions(currentRequest.sessionData)
fault.retries = 0 return applyPresets(currentRequest)
currentRequest.sessionData.faults[channel] = fault
util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})")
return nextRpc(currentRequest) if not rpcRequest?
for channel, flags of currentRequest.sessionData.channels
if flags and channel of currentRequest.sessionData.faults
delete currentRequest.sessionData.faults[channel]
currentRequest.sessionData.faultsTouched ?= {}
currentRequest.sessionData.faultsTouched[channel] = true
for p, i in doneProvisions or [] if whiteList?
delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]] return applyPresets(currentRequest)
sendRpcRequest(currentRequest, id, rpcRequest) sendRpcRequest(currentRequest, id, rpcRequest)
) )
@@ -252,11 +328,11 @@ applyPresets = (currentRequest) ->
nextRpc = (currentRequest) -> nextRpc = (currentRequest) ->
session.rpcRequest(currentRequest.sessionData, null, (err, faults, id, rpcRequest, doneProvisions) -> session.rpcRequest(currentRequest.sessionData, null, (err, fault, id, rpcRequest) ->
throw err if err throw err if err
if faults? if fault
for channel, fault of faults for channel of currentRequest.sessionData.channels
if channel.startsWith('_task_') if channel.startsWith('_task_')
taskId = channel.slice(6) taskId = channel.slice(6)
for t, j in currentRequest.sessionData.tasks for t, j in currentRequest.sessionData.tasks
@@ -265,19 +341,20 @@ nextRpc = (currentRequest) ->
fault.expiry = t.expiry if t.expiry? fault.expiry = t.expiry if t.expiry?
break break
if not fault.retries? and Object.keys(faults).length == 1 recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels)
fault.retries = 0 session.clearProvisions(currentRequest.sessionData)
currentRequest.sessionData.faults[channel] = fault
util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})")
return nextRpc(currentRequest) return nextRpc(currentRequest)
if rpcRequest? if rpcRequest?
return sendRpcRequest(currentRequest, id, rpcRequest) return sendRpcRequest(currentRequest, id, rpcRequest)
for p, i in doneProvisions or [] for channel, flags of currentRequest.sessionData.channels
delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]] if flags and channel of currentRequest.sessionData.faults
delete currentRequest.sessionData.faults[channel]
currentRequest.sessionData.faultsTouched ?= {}
currentRequest.sessionData.faultsTouched[channel] = true
for p in currentRequest.sessionData.provisions
if p[0] == '_task' if p[0] == '_task'
currentRequest.sessionData.doneTasks ?= [] currentRequest.sessionData.doneTasks ?= []
currentRequest.sessionData.doneTasks.push(p[1]) currentRequest.sessionData.doneTasks.push(p[1])
@@ -286,6 +363,8 @@ nextRpc = (currentRequest) ->
currentRequest.sessionData.tasks.splice(j, 1) currentRequest.sessionData.tasks.splice(j, 1)
break break
session.clearProvisions(currentRequest.sessionData)
for task in currentRequest.sessionData.tasks for task in currentRequest.sessionData.tasks
# Delete if expired # Delete if expired
if task.expiry <= currentRequest.sessionData.timestamp if task.expiry <= currentRequest.sessionData.timestamp
@@ -333,13 +412,32 @@ sendRpcRequest = (currentRequest, id, rpcRequest) ->
db.clearTasks(currentRequest.sessionData.doneTasks, (err) -> db.clearTasks(currentRequest.sessionData.doneTasks, (err) ->
throw err if err throw err if err
if not currentRequest.sessionData.faults? counter = 1
return writeResponse(currentRequest, soap.response(null))
db.syncFaults(currentRequest.sessionData.deviceId, currentRequest.sessionData.faults, (err) -> for k of currentRequest.sessionData.faultsTouched
throw err if err ++ counter
writeResponse(currentRequest, soap.response(null)) if currentRequest.sessionData.faults[k]
) currentRequest.sessionData.faults[k].retries = currentRequest.sessionData.retries[k]
db.saveFault(currentRequest.sessionData.deviceId, k, currentRequest.sessionData.faults[k], (err) ->
-- counter
if err and counter >= 0
throw err
if counter == 0
return writeResponse(currentRequest, soap.response(null))
)
else
db.deleteFault(currentRequest.sessionData.deviceId, k, (err) ->
-- counter
if err and counter >= 0
throw err
if counter == 0
return writeResponse(currentRequest, soap.response(null))
)
if -- counter == 0
return writeResponse(currentRequest, soap.response(null))
) )
) )
return return
@@ -490,7 +588,10 @@ listener = (httpRequest, httpResponse) ->
# Delete expired faults # Delete expired faults
for k, v of sessionData.faults for k, v of sessionData.faults
delete sessionData.faults[k] if v.expiry >= sessionData.timestamp if v.expiry >= sessionData.timestamp
delete sessionData.faults[k]
sessionData.faultsTouched ?= {}
sessionData.faultsTouched[k] = true
f(null, sessionData) f(null, sessionData)
) )
@@ -529,36 +630,32 @@ listener = (httpRequest, httpResponse) ->
nextRpc(currentRequest) nextRpc(currentRequest)
) )
else if cwmpRequest.fault? else if cwmpRequest.fault?
session.rpcFault(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.fault, (err, faults) -> session.rpcFault(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.fault, (err, fault) ->
throw err if err throw err if err
for channel, fault of faults if fault
if channel.startsWith('_task_') for channel of currentRequest.sessionData.channels
taskId = channel.slice(6) if channel.startsWith('_task_')
for t, j in currentRequest.sessionData.tasks taskId = channel.slice(6)
if t._id == taskId for t, j in currentRequest.sessionData.tasks
currentRequest.sessionData.tasks.splice(j, 1) if t._id == taskId
fault.expiry = t.expiry if t.expiry? currentRequest.sessionData.tasks.splice(j, 1)
break fault.expiry = t.expiry if t.expiry?
break
if not fault.retries? and Object.keys(faults).length == 1
fault.retries = 0
currentRequest.sessionData.faults[channel] = fault
util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})")
recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels)
session.clearProvisions(currentRequest.sessionData)
nextRpc(currentRequest) nextRpc(currentRequest)
) )
else # CPE sent empty response else # CPE sent empty response
session.timeoutOperations(currentRequest.sessionData, (err, faults) -> session.timeoutOperations(currentRequest.sessionData, (err, faults, operations) ->
throw err if err throw err if err
for channel, fault of faults for fault, i in faults
if not fault.retries? and Object.keys(faults).length == 1 for k, v of operations[i].retries
fault.retries = 0 sessionData.retries[k] = v
currentRequest.sessionData.faults[channel] = fault recordFault(currentRequest.sessionData, fault, operations[i].provisions, operations[i].channels)
util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})")
nextRpc(currentRequest) nextRpc(currentRequest)
) )
+23 -56
View File
@@ -527,6 +527,7 @@ getFaults = (deviceId, callback) ->
delete r.device delete r.device
r.timestamp = +r.timestamp r.timestamp = +r.timestamp
r.expiry = +r.expiry if r.expiry? r.expiry = +r.expiry if r.expiry?
r.provisions = JSON.parse(r.provisions)
faults[channel] = r faults[channel] = r
CACHE_DURATION = config.get('PRESETS_CACHE_DURATION', deviceId) CACHE_DURATION = config.get('PRESETS_CACHE_DURATION', deviceId)
@@ -537,6 +538,25 @@ getFaults = (deviceId, callback) ->
) )
saveFault = (deviceId, channel, fault, callback) ->
fault._id = "#{deviceId}:#{channel}"
fault.device = deviceId
fault.channel = channel
fault.timestamp = new Date(fault.timestamp)
fault.provisions = JSON.stringify(fault.provisions)
faultsCollection.save(fault, (err) ->
return callback(err) if err
redisClient.del("#{deviceId}_faults", callback)
)
deleteFault = (deviceId, channel, callback) ->
faultsCollection.remove({_id: "#{deviceId}:#{channel}"}, (err) ->
return callback(err) if err
redisClient.del("#{deviceId}_faults", callback)
)
getDueTasks = (deviceId, timestamp, callback) -> getDueTasks = (deviceId, timestamp, callback) ->
cur = tasksCollection.find({ cur = tasksCollection.find({
@@ -586,61 +606,6 @@ clearTasks = (taskIds, callback) ->
tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, callback) tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, callback)
syncFaults = (deviceId, faults, callback) ->
getFaults(deviceId, (err, existingFaults) ->
return callback(err) if err
counter = 1
toDelete = []
for k of existingFaults
if k not of faults
toDelete.push("#{deviceId}:#{k}")
if toDelete.length
++ counter
faultsCollection.remove({'_id' : {'$in' : toDelete}}, (err) ->
if err
callback(err) if counter
return counter = 0
if -- counter == 0
return redisClient.del("#{deviceId}_faults", callback)
)
for channel, fault of faults
continue if existingFaults[channel]? and fault.retries == existingFaults[channel].retries
do (channel, fault) ->
++ counter
fault._id = "#{deviceId}:#{channel}"
fault.device = deviceId
fault.channel = channel
fault.timestamp = new Date(fault.timestamp)
fault.expiry = new Date(fault.expiry) if fault.expiry?
faultsCollection.save(fault, (err) ->
if err
callback(err) if counter
return counter = 0
if channel.startsWith('_task_')
return tasksCollection.update({_id: new mongodb.ObjectID(channel.slice(6))}, {$set: {fault: fault.fault, retries: fault.retries}}, (err) ->
if err
callback(err) if counter
return counter = 0
if -- counter == 0
return redisClient.del("#{deviceId}_faults", callback)
)
else
if -- counter == 0
return redisClient.del("#{deviceId}_faults", callback)
)
if -- counter == 0
return callback()
)
getOperations = (deviceId, callback) -> getOperations = (deviceId, callback) ->
operationsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) -> operationsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) ->
return callback(err) if err return callback(err) if err
@@ -687,7 +652,9 @@ exports.disconnect = disconnect
exports.fetchDevice = fetchDevice exports.fetchDevice = fetchDevice
exports.saveDevice = saveDevice exports.saveDevice = saveDevice
exports.getDueTasksAndFaultsAndOperations = getDueTasksAndFaultsAndOperations exports.getDueTasksAndFaultsAndOperations = getDueTasksAndFaultsAndOperations
exports.getFaults = getFaults
exports.saveFault = saveFault
exports.deleteFault = deleteFault
exports.clearTasks = clearTasks exports.clearTasks = clearTasks
exports.syncFaults = syncFaults
exports.saveOperation = saveOperation exports.saveOperation = saveOperation
exports.deleteOperation = deleteOperation exports.deleteOperation = deleteOperation
+84 -134
View File
@@ -47,8 +47,9 @@ init = (deviceId, cwmpVersion, timeout, callback) ->
cwmpVersion : cwmpVersion cwmpVersion : cwmpVersion
timeout : timeout timeout : timeout
provisions: [] provisions: []
channels: [] channels: {}
revisions: [] virtualParameters: []
revisions: [0]
rpcCount: 0 rpcCount: 0
iteration: 0 iteration: 0
extensionsCache: {} extensionsCache: {}
@@ -159,7 +160,6 @@ transferComplete = (sessionData, rpcReq, callback) ->
# TODO Show a warning instead # TODO Show a warning instead
throw new Error('Invalid command key') throw new Error('Invalid command key')
instance = operation.args.instance instance = operation.args.instance
faults = {}
delete sessionData.operations[commandKey] delete sessionData.operations[commandKey]
sessionData.operationsTouched ?= {} sessionData.operationsTouched ?= {}
@@ -167,19 +167,14 @@ transferComplete = (sessionData, rpcReq, callback) ->
if rpcReq.faultStruct? and rpcReq.faultStruct.faultCode != '0' if rpcReq.faultStruct? and rpcReq.faultStruct.faultCode != '0'
return revertDownloadParameters(sessionData, operation.args.instance, (err) -> return revertDownloadParameters(sessionData, operation.args.instance, (err) ->
for channel, provisions of operation.provisions fault = {
faults[channel] = { code: "cwmp.#{rpcReq.faultStruct.faultCode}"
provisions: provisions message: rpcReq.faultStruct.faultString
timestamp: sessionData.timestamp detail: rpcReq.faultStruct
code: "cwmp.#{rpcReq.faultStruct.faultCode}" timestamp: operation.timestamp
message: rpcReq.faultStruct.faultString }
detail: rpcReq.faultStruct
}
if operation.retries[channel]? return callback(err, {type : 'TransferCompleteResponse'}, fault, operation)
faults[channel].retries = operation.retries[channel] + 1
return callback(err, {type : 'TransferCompleteResponse'}, faults)
) )
loadPath(sessionData, ['Downloads', instance, '*']) loadPath(sessionData, ['Downloads', instance, '*'])
@@ -215,7 +210,7 @@ transferComplete = (sessionData, rpcReq, callback) ->
{value: [timestamp, [+rpcReq.completeTime, 'xsd:dateTime']]}, toClear) {value: [timestamp, [+rpcReq.completeTime, 'xsd:dateTime']]}, toClear)
clear(sessionData, toClear, (err) -> clear(sessionData, toClear, (err) ->
return callback(err, {type : 'TransferCompleteResponse'}, faults) return callback(err, {type : 'TransferCompleteResponse'})
) )
) )
@@ -240,7 +235,8 @@ revertDownloadParameters = (sessionData, instance, callback) ->
timeoutOperations = (sessionData, callback) -> timeoutOperations = (sessionData, callback) ->
faults = {} faults = []
operations = []
counter = 1 counter = 1
for commandKey, operation of sessionData.operations for commandKey, operation of sessionData.operations
@@ -254,16 +250,13 @@ timeoutOperations = (sessionData, callback) ->
sessionData.operationsTouched ?= {} sessionData.operationsTouched ?= {}
sessionData.operationsTouched[commandKey] = 1 sessionData.operationsTouched[commandKey] = 1
for channel, provisions of operation.provisions faults.push({
faults[channel] = { code: 'timeout'
provisions: provisions message: 'Download operation timed out'
timestamp: sessionData.timestamp timestamp: operation.timestamp
code: 'timeout' })
message: 'Download operation timeed out'
}
if operation.retries[channel]? operations.push(operation)
faults[channel].retries = operation.retries[channel] + 1
++ counter ++ counter
revertDownloadParameters(sessionData, operation.args.instance, (err) -> revertDownloadParameters(sessionData, operation.args.instance, (err) ->
@@ -271,36 +264,41 @@ timeoutOperations = (sessionData, callback) ->
if err and counter > 0 if err and counter > 0
counter = 0 counter = 0
return callback(err, faults) if counter == 0 return callback(err, faults, operations) if counter == 0
) )
-- counter -- counter
return callback(null, faults) if counter == 0 return callback(null, faults, operations) if counter == 0
addProvisions = (sessionData, channel, provisions) -> addProvisions = (sessionData, channel, provisions) ->
if sessionData.revisions[0] > 0
sessionData.deviceData.timestamps.collapse(1)
sessionData.deviceData.attributes.collapse(1)
delete sessionData.syncState delete sessionData.syncState
delete sessionData.rpcRequest delete sessionData.rpcRequest
sessionData.declarations = [] sessionData.declarations = []
sessionData.doneProvisions = 0 sessionData.doneProvisions = 0
sessionData.provisions[0] ?= [] if sessionData.revisions[sessionData.revisions.length - 1] > 0
sessionData.provisions.length = 1 sessionData.deviceData.timestamps.collapse(1)
sessionData.revisions = [0] sessionData.deviceData.attributes.collapse(1)
sessionData.extensionsCache = {} sessionData.revisions = [0]
sessionData.extensionsCache = {}
channels = [channel]
for provision, i in provisions for provision, i in provisions
# Remove duplicate provisions # Remove duplicate provisions
for p, j in sessionData.provisions[0] provisionStr = JSON.stringify(provision)
if channel == sessionData.channels[j] and JSON.stringify(p) == JSON.stringify(provision) for p, j in sessionData.provisions
sessionData.provisions[0].splice(j, 1) if JSON.stringify(p) == provisionStr
sessionData.channels.splice(j, 1) sessionData.provisions.splice(j, 1)
for c of sessionData.channels
channels.push(c) if sessionData.channels[c] & (1 << j)
a = sessionData.channels[c] >> (j + 1)
sessionData.channels[c] &= ((1 << j) - 1)
sessionData.channels[c] |= (a << j)
sessionData.provisions[0].push(provision) for c in channels
sessionData.channels.push(channel) sessionData.channels[channel] |= (1 << sessionData.provisions.length)
sessionData.provisions.push(provision)
clearProvisions = (sessionData) -> clearProvisions = (sessionData) ->
@@ -311,10 +309,10 @@ clearProvisions = (sessionData) ->
delete sessionData.syncState delete sessionData.syncState
delete sessionData.rpcRequest delete sessionData.rpcRequest
sessionData.provisions = [] sessionData.provisions = []
sessionData.channels = [] sessionData.channels = {}
sessionData.declarations = [] sessionData.declarations = []
sessionData.doneProvisions = 0 sessionData.doneProvisions = 0
sessionData.revisions = [] sessionData.revisions = [0]
sessionData.extensionsCache = {} sessionData.extensionsCache = {}
@@ -584,39 +582,32 @@ rpcRequest = (sessionData, _declarations, callback) ->
if sessionData.rpcRequest? if sessionData.rpcRequest?
return callback(null, null, generateRpcId(sessionData), sessionData.rpcRequest) return callback(null, null, generateRpcId(sessionData), sessionData.rpcRequest)
if sessionData.declarations.length < sessionData.provisions.length if sessionData.virtualParameters.length == 0 and
sessionData.declarations.length == 0 and
sessionData.doneProvisions
return callback()
if sessionData.declarations.length <= sessionData.virtualParameters.length
inception = sessionData.declarations.length inception = sessionData.declarations.length
revision = sessionData.revisions[inception] + 1 revision = sessionData.revisions[inception] + 1
sessionData.deviceData.timestamps.revision = revision sessionData.deviceData.timestamps.revision = revision
sessionData.deviceData.attributes.revision = revision sessionData.deviceData.attributes.revision = revision
run = if inception == 0 then runProvisions else runVirtualParameters
return run(sessionData, sessionData.provisions[inception], sessionData.revisions[inception - 1] ? 0, sessionData.revisions[inception], (err, fault, done, decs, toClear) -> if inception == 0
run = runProvisions
provisions = sessionData.provisions
else
run = runVirtualParameters
provisions = sessionData.virtualParameters[inception - 1]
return run(sessionData, provisions, sessionData.revisions[inception - 1] or 0, sessionData.revisions[inception], (err, fault, done, decs, toClear) ->
return callback(err) if err return callback(err) if err
if fault if fault
faults = {} fault.timestamp = sessionData.timestamp
for p, i in sessionData.provisions[0] fault.provisions = sessionData.provisions
channel = sessionData.channels[i] fault.channels = sessionData.channels
f = faults[channel] return callback(null, fault)
if not f?
f = {
provisions: []
timestamp: sessionData.timestamp
code: fault.code
message: fault.message
detail: fault.detail
}
if sessionData.retries[channel]?
f.retries = sessionData.retries[channel] + 1
faults[channel] = f
f.provisions.push(p)
clearProvisions(sessionData)
return callback(null, faults)
sessionData.declarations.push(decs) sessionData.declarations.push(decs)
sessionData.doneProvisions |= 1 << inception if done or not decs.length sessionData.doneProvisions |= 1 << inception if done or not decs.length
@@ -638,8 +629,6 @@ rpcRequest = (sessionData, _declarations, callback) ->
delete sessionData.syncState delete sessionData.syncState
sessionData.declarations[0] ?= [] sessionData.declarations[0] ?= []
sessionData.declarations[0] = sessionData.declarations[0].concat(_declarations) sessionData.declarations[0] = sessionData.declarations[0].concat(_declarations)
sessionData.provisions[0] ?= []
sessionData.revisions[0] ?= 0
for d in _declarations for d in _declarations
for ad in device.getAliasDeclarations(d[0], 1) for ad in device.getAliasDeclarations(d[0], 1)
@@ -687,27 +676,12 @@ rpcRequest = (sessionData, _declarations, callback) ->
sessionData.revisions.length >= 8 or sessionData.revisions.length >= 8 or
sessionData.iteration >= 64 sessionData.iteration >= 64
faults = {} fault = {
for p, i in sessionData.provisions[0] code: 'endless_cycle'
channel = sessionData.channels[i] message: 'The provision seems to be repeating indefinitely'
fault = faults[channel] timestmap: sessionData.timestamp
if not fault? }
fault = { return callback(null, fault)
provisions: []
timestamp: sessionData.timestamp
code: 'endless_cycle'
message: 'The provision seems to be repeating indefinitely'
}
if sessionData.retries[channel]?
fault.retries = sessionData.retries[channel] + 1
faults[channel] = fault
fault.provisions.push(p)
clearProvisions(sessionData)
return callback(null, faults)
inception = sessionData.declarations.length - 1 inception = sessionData.declarations.length - 1
@@ -818,7 +792,7 @@ rpcRequest = (sessionData, _declarations, callback) ->
sessionData.rpcRequest = generateSetRpcRequest(sessionData) sessionData.rpcRequest = generateSetRpcRequest(sessionData)
if provisions if provisions
sessionData.provisions.push(provisions) sessionData.virtualParameters.push(provisions)
sessionData.revisions.push(sessionData.revisions[inception]) sessionData.revisions.push(sessionData.revisions[inception])
return rpcRequest(sessionData, null, callback) return rpcRequest(sessionData, null, callback)
@@ -827,18 +801,12 @@ rpcRequest = (sessionData, _declarations, callback) ->
++ sessionData.revisions[inception] ++ sessionData.revisions[inception]
sessionData.declarations.pop() sessionData.declarations.pop()
while sessionData.doneProvisions & (1 << (sessionData.provisions.length - 1))
doneProvisions = sessionData.provisions.pop() while sessionData.doneProvisions & (2 << (sessionData.virtualParameters.length - 1))
doneProvisions = sessionData.virtualParameters.pop()
sessionData.revisions.pop() sessionData.revisions.pop()
sessionData.declarations.pop() sessionData.declarations.pop()
sessionData.doneProvisions &= (1 << sessionData.provisions.length) - 1 sessionData.doneProvisions &= (2 << sessionData.virtualParameters.length) - 1
if not sessionData.provisions.length
delete sessionData.syncState
sessionData.deviceData.timestamps.collapse(1)
sessionData.deviceData.attributes.collapse(1)
sessionData.extensionsCache = {}
return callback(null, null, null, null, doneProvisions)
rev = sessionData.revisions[sessionData.revisions.length - 1] rev = sessionData.revisions[sessionData.revisions.length - 1]
sessionData.deviceData.timestamps.collapse(rev + 1) sessionData.deviceData.timestamps.collapse(rev + 1)
sessionData.deviceData.attributes.collapse(rev + 1) sessionData.deviceData.attributes.collapse(rev + 1)
@@ -1527,7 +1495,8 @@ rpcResponse = (sessionData, id, rpcRes, callback) ->
operation = { operation = {
type: 'Download' type: 'Download'
timestamp: sessionData.timestamp timestamp: sessionData.timestamp
provisions: {} provisions: sessionData.provisions
channels: sessionData.channels
retries: {} retries: {}
args: { args: {
instance: rpcReq.instance instance: rpcReq.instance
@@ -1537,12 +1506,9 @@ rpcResponse = (sessionData, id, rpcRes, callback) ->
} }
} }
for provision, i in sessionData.provisions[0] for channel of sessionData.channels
channel = sessionData.channels[i] if sessionData.retries[channel]?
operation.provisions[channel] ?= [] operation.retries[channel] = sessionData.retries[channel]
operation.provisions[channel].push(provision)
if sessionData.faults[channel]?
operation.retries[channel] = sessionData.faults[channel].retries or 0
sessionData.operations[rpcReq.commandKey] = operation sessionData.operations[rpcReq.commandKey] = operation
sessionData.operationsTouched ?= {} sessionData.operationsTouched ?= {}
@@ -1558,31 +1524,15 @@ rpcFault = (sessionData, id, faultResponse, callback) ->
# TODO Consider handling invalid parameter faults by automatically refreshing # TODO Consider handling invalid parameter faults by automatically refreshing
# relevant data model portions # relevant data model portions
if not sessionData.provisions[0]?.length fault = {
throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(flt)) code: "cwmp.#{faultResponse.detail.faultCode}"
message: faultResponse.detail.faultString
detail: faultResponse.detail
timestamp: sessionData.timestamp
}
faults = {} delete sessionData.syncState
for p, i in sessionData.provisions[0] return callback(null, fault)
channel = sessionData.channels[i]
fault = faults[channel]
if not fault?
fault = {
provisions: []
timestamp: sessionData.timestamp
code: "cwmp.#{faultResponse.detail.faultCode}"
message: faultResponse.detail.faultString
detail: faultResponse.detail
}
if sessionData.retries[channel]?
fault.retries = sessionData.retries[channel] + 1
faults[channel] = fault
fault.provisions.push(p)
clearProvisions(sessionData)
return callback(null, faults)
deserialize = (sessionDataString, callback) -> deserialize = (sessionDataString, callback) ->