From 63407be19bb0b1a679214d97cf2c50a2e5adc977 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Tue, 20 Dec 2016 16:23:43 -0800 Subject: [PATCH] Generate faults in precondition testing stage Also, queue applicable provisions to any existing fault in order to be processed later. --- lib/cwmp.coffee | 241 +++++++++++++++++++++++++++++++-------------- lib/db.coffee | 79 +++++---------- lib/session.coffee | 218 ++++++++++++++++------------------------ 3 files changed, 276 insertions(+), 262 deletions(-) diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index bea5450..02f4f00 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -83,6 +83,41 @@ writeResponse = (currentRequest, res) -> 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) -> if config.get('LOG_INFORMS', currentRequest.sessionData.deviceId) util.log("#{currentRequest.sessionData.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") @@ -105,15 +140,14 @@ inform = (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 - for channel, fault of faults - if not fault.retries? and Object.keys(faults).length == 1 - fault.retries = 0 + if fault + for k, v of operation.retries + currentRequest.sessionData.retries[k] = v - currentRequest.sessionData.faults[channel] = fault - util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})") + recordFault(currentRequest.sessionData, fault, operation.provisions, operation.channels) res = soap.response({ id : cwmpRequest.id, @@ -146,12 +180,37 @@ testSchedule = (schedule, 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) -> cache.getPresets((err, presetsHash, presets) -> throw err if err # Filter presets based on existing faults - blackList = null + blackList = {} whiteList = null whiteListProvisions = null @@ -159,18 +218,17 @@ applyPresets = (currentRequest) -> if currentRequest.sessionData.faults? for channel, fault of currentRequest.sessionData.faults - if fault.retries? - retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, fault.retries)) * 1000 - else + if fault.retryNow retryTimestamp = 0 + else + retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, currentRequest.sessionData.retries[channel])) * 1000 if retryTimestamp <= currentRequest.sessionData.timestamp whiteList = channel whiteListProvisions = fault.provisions break - blackList ?= [] - blackList.push(channel) + blackList[channel] = if fault.precondition then 1 else 2 currentRequest.sessionData.presetsHash = presetsHash @@ -183,23 +241,24 @@ applyPresets = (currentRequest) -> parameters = {} filteredPresets = [] - for p in presets + for preset in presets if whiteList? - continue if p.channel != whiteList - else if blackList and p.channel in blackList + continue if preset.channel != whiteList + else if blackList[channel] == 1 continue eventsMatch = true - for k, v of p.events + for k, v of preset.events if (!v) != (!deviceEvents[k.replace(' ', '_')]) eventsMatch = false break - continue if (not eventsMatch) or (p.schedule and not - (p.schedule.schedule and testSchedule(p.schedule, currentRequest.sessionData.timestamp)[0])) + continue if (not eventsMatch) or (preset.schedule and not + (preset.schedule.schedule and testSchedule(preset.schedule, currentRequest.sessionData.timestamp)[0])) - filteredPresets.push(p) - for k of p.precondition + filteredPresets.push(preset) + for k of preset.precondition + currentRequest.sessionData.channels[preset.channel] = 0 p = k.split(/([^a-zA-Z0-9\-\_\.].*)/, 1)[0] parameters[p] = common.parsePath(p) @@ -207,15 +266,19 @@ applyPresets = (currentRequest) -> for k, v of parameters 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 - if faults? - throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(faults)) + if fault + recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels) + session.clearProvisions(currentRequest.sessionData) + return applyPresets(currentRequest) if rpcRequest? return sendRpcRequest(currentRequest, id, rpcRequest) + session.clearProvisions(currentRequest.sessionData) + for k, v of parameters unpacked = device.unpack(currentRequest.sessionData.deviceData, v) if unpacked[0] and (vv = currentRequest.sessionData.deviceData.attributes.get(unpacked[0]).value?[1])? @@ -226,24 +289,37 @@ applyPresets = (currentRequest) -> if whiteList? session.addProvisions(currentRequest.sessionData, whiteList, whiteListProvisions) + appendProvisionsToFaults = {} for p in filteredPresets 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 - if faults? - for channel, fault of faults - 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})") + if fault + recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels) + session.clearProvisions(currentRequest.sessionData) + return applyPresets(currentRequest) - 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 [] - delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]] + if whiteList? + return applyPresets(currentRequest) sendRpcRequest(currentRequest, id, rpcRequest) ) @@ -252,11 +328,11 @@ applyPresets = (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 - if faults? - for channel, fault of faults + if fault + for channel of currentRequest.sessionData.channels if channel.startsWith('_task_') taskId = channel.slice(6) for t, j in currentRequest.sessionData.tasks @@ -265,19 +341,20 @@ nextRpc = (currentRequest) -> 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) return nextRpc(currentRequest) if rpcRequest? return sendRpcRequest(currentRequest, id, rpcRequest) - for p, i in doneProvisions or [] - delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]] + 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 in currentRequest.sessionData.provisions if p[0] == '_task' currentRequest.sessionData.doneTasks ?= [] currentRequest.sessionData.doneTasks.push(p[1]) @@ -286,6 +363,8 @@ nextRpc = (currentRequest) -> currentRequest.sessionData.tasks.splice(j, 1) break + session.clearProvisions(currentRequest.sessionData) + for task in currentRequest.sessionData.tasks # Delete if expired if task.expiry <= currentRequest.sessionData.timestamp @@ -333,13 +412,32 @@ sendRpcRequest = (currentRequest, id, rpcRequest) -> db.clearTasks(currentRequest.sessionData.doneTasks, (err) -> throw err if err - if not currentRequest.sessionData.faults? - return writeResponse(currentRequest, soap.response(null)) + counter = 1 - db.syncFaults(currentRequest.sessionData.deviceId, currentRequest.sessionData.faults, (err) -> - throw err if err - writeResponse(currentRequest, soap.response(null)) - ) + for k of currentRequest.sessionData.faultsTouched + ++ counter + 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 @@ -490,7 +588,10 @@ listener = (httpRequest, httpResponse) -> # Delete expired 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) ) @@ -529,36 +630,32 @@ listener = (httpRequest, httpResponse) -> nextRpc(currentRequest) ) 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 - for channel, fault of faults - if channel.startsWith('_task_') - taskId = channel.slice(6) - for t, j in currentRequest.sessionData.tasks - if t._id == taskId - currentRequest.sessionData.tasks.splice(j, 1) - 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})") + if fault + for channel of currentRequest.sessionData.channels + if channel.startsWith('_task_') + taskId = channel.slice(6) + for t, j in currentRequest.sessionData.tasks + if t._id == taskId + currentRequest.sessionData.tasks.splice(j, 1) + fault.expiry = t.expiry if t.expiry? + break + recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels) + session.clearProvisions(currentRequest.sessionData) nextRpc(currentRequest) ) else # CPE sent empty response - session.timeoutOperations(currentRequest.sessionData, (err, faults) -> + session.timeoutOperations(currentRequest.sessionData, (err, faults, operations) -> throw err if err - for channel, fault of faults - if not fault.retries? and Object.keys(faults).length == 1 - fault.retries = 0 + for fault, i in faults + for k, v of operations[i].retries + sessionData.retries[k] = v - currentRequest.sessionData.faults[channel] = fault - util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})") + recordFault(currentRequest.sessionData, fault, operations[i].provisions, operations[i].channels) nextRpc(currentRequest) ) diff --git a/lib/db.coffee b/lib/db.coffee index ce640f7..0484239 100644 --- a/lib/db.coffee +++ b/lib/db.coffee @@ -527,6 +527,7 @@ getFaults = (deviceId, callback) -> delete r.device r.timestamp = +r.timestamp r.expiry = +r.expiry if r.expiry? + r.provisions = JSON.parse(r.provisions) faults[channel] = r 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) -> cur = tasksCollection.find({ @@ -586,61 +606,6 @@ clearTasks = (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) -> operationsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) -> return callback(err) if err @@ -687,7 +652,9 @@ exports.disconnect = disconnect exports.fetchDevice = fetchDevice exports.saveDevice = saveDevice exports.getDueTasksAndFaultsAndOperations = getDueTasksAndFaultsAndOperations +exports.getFaults = getFaults +exports.saveFault = saveFault +exports.deleteFault = deleteFault exports.clearTasks = clearTasks -exports.syncFaults = syncFaults exports.saveOperation = saveOperation exports.deleteOperation = deleteOperation diff --git a/lib/session.coffee b/lib/session.coffee index cdcda3b..4943856 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -47,8 +47,9 @@ init = (deviceId, cwmpVersion, timeout, callback) -> cwmpVersion : cwmpVersion timeout : timeout provisions: [] - channels: [] - revisions: [] + channels: {} + virtualParameters: [] + revisions: [0] rpcCount: 0 iteration: 0 extensionsCache: {} @@ -159,7 +160,6 @@ transferComplete = (sessionData, rpcReq, callback) -> # TODO Show a warning instead throw new Error('Invalid command key') instance = operation.args.instance - faults = {} delete sessionData.operations[commandKey] sessionData.operationsTouched ?= {} @@ -167,19 +167,14 @@ transferComplete = (sessionData, rpcReq, callback) -> if rpcReq.faultStruct? and rpcReq.faultStruct.faultCode != '0' return revertDownloadParameters(sessionData, operation.args.instance, (err) -> - for channel, provisions of operation.provisions - faults[channel] = { - provisions: provisions - timestamp: sessionData.timestamp - code: "cwmp.#{rpcReq.faultStruct.faultCode}" - message: rpcReq.faultStruct.faultString - detail: rpcReq.faultStruct - } + fault = { + code: "cwmp.#{rpcReq.faultStruct.faultCode}" + message: rpcReq.faultStruct.faultString + detail: rpcReq.faultStruct + timestamp: operation.timestamp + } - if operation.retries[channel]? - faults[channel].retries = operation.retries[channel] + 1 - - return callback(err, {type : 'TransferCompleteResponse'}, faults) + return callback(err, {type : 'TransferCompleteResponse'}, fault, operation) ) loadPath(sessionData, ['Downloads', instance, '*']) @@ -215,7 +210,7 @@ transferComplete = (sessionData, rpcReq, callback) -> {value: [timestamp, [+rpcReq.completeTime, 'xsd:dateTime']]}, toClear) 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) -> - faults = {} + faults = [] + operations = [] counter = 1 for commandKey, operation of sessionData.operations @@ -254,16 +250,13 @@ timeoutOperations = (sessionData, callback) -> sessionData.operationsTouched ?= {} sessionData.operationsTouched[commandKey] = 1 - for channel, provisions of operation.provisions - faults[channel] = { - provisions: provisions - timestamp: sessionData.timestamp - code: 'timeout' - message: 'Download operation timeed out' - } + faults.push({ + code: 'timeout' + message: 'Download operation timed out' + timestamp: operation.timestamp + }) - if operation.retries[channel]? - faults[channel].retries = operation.retries[channel] + 1 + operations.push(operation) ++ counter revertDownloadParameters(sessionData, operation.args.instance, (err) -> @@ -271,36 +264,41 @@ timeoutOperations = (sessionData, callback) -> if err and counter > 0 counter = 0 - return callback(err, faults) if counter == 0 + return callback(err, faults, operations) if counter == 0 ) -- counter - return callback(null, faults) if counter == 0 + return callback(null, faults, operations) if counter == 0 addProvisions = (sessionData, channel, provisions) -> - if sessionData.revisions[0] > 0 - sessionData.deviceData.timestamps.collapse(1) - sessionData.deviceData.attributes.collapse(1) - delete sessionData.syncState delete sessionData.rpcRequest sessionData.declarations = [] sessionData.doneProvisions = 0 - sessionData.provisions[0] ?= [] - sessionData.provisions.length = 1 - sessionData.revisions = [0] - sessionData.extensionsCache = {} + if sessionData.revisions[sessionData.revisions.length - 1] > 0 + sessionData.deviceData.timestamps.collapse(1) + sessionData.deviceData.attributes.collapse(1) + sessionData.revisions = [0] + sessionData.extensionsCache = {} + + channels = [channel] for provision, i in provisions # Remove duplicate provisions - for p, j in sessionData.provisions[0] - if channel == sessionData.channels[j] and JSON.stringify(p) == JSON.stringify(provision) - sessionData.provisions[0].splice(j, 1) - sessionData.channels.splice(j, 1) + provisionStr = JSON.stringify(provision) + for p, j in sessionData.provisions + if JSON.stringify(p) == provisionStr + 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) - sessionData.channels.push(channel) + for c in channels + sessionData.channels[channel] |= (1 << sessionData.provisions.length) + sessionData.provisions.push(provision) clearProvisions = (sessionData) -> @@ -311,10 +309,10 @@ clearProvisions = (sessionData) -> delete sessionData.syncState delete sessionData.rpcRequest sessionData.provisions = [] - sessionData.channels = [] + sessionData.channels = {} sessionData.declarations = [] sessionData.doneProvisions = 0 - sessionData.revisions = [] + sessionData.revisions = [0] sessionData.extensionsCache = {} @@ -584,39 +582,32 @@ rpcRequest = (sessionData, _declarations, callback) -> if 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 revision = sessionData.revisions[inception] + 1 sessionData.deviceData.timestamps.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 if fault - faults = {} - for p, i in sessionData.provisions[0] - channel = sessionData.channels[i] - f = faults[channel] - 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) + fault.timestamp = sessionData.timestamp + fault.provisions = sessionData.provisions + fault.channels = sessionData.channels + return callback(null, fault) sessionData.declarations.push(decs) sessionData.doneProvisions |= 1 << inception if done or not decs.length @@ -638,8 +629,6 @@ rpcRequest = (sessionData, _declarations, callback) -> delete sessionData.syncState sessionData.declarations[0] ?= [] sessionData.declarations[0] = sessionData.declarations[0].concat(_declarations) - sessionData.provisions[0] ?= [] - sessionData.revisions[0] ?= 0 for d in _declarations for ad in device.getAliasDeclarations(d[0], 1) @@ -687,27 +676,12 @@ rpcRequest = (sessionData, _declarations, callback) -> sessionData.revisions.length >= 8 or sessionData.iteration >= 64 - faults = {} - for p, i in sessionData.provisions[0] - channel = sessionData.channels[i] - fault = faults[channel] - if not fault? - 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) + fault = { + code: 'endless_cycle' + message: 'The provision seems to be repeating indefinitely' + timestmap: sessionData.timestamp + } + return callback(null, fault) inception = sessionData.declarations.length - 1 @@ -818,7 +792,7 @@ rpcRequest = (sessionData, _declarations, callback) -> sessionData.rpcRequest = generateSetRpcRequest(sessionData) if provisions - sessionData.provisions.push(provisions) + sessionData.virtualParameters.push(provisions) sessionData.revisions.push(sessionData.revisions[inception]) return rpcRequest(sessionData, null, callback) @@ -827,18 +801,12 @@ rpcRequest = (sessionData, _declarations, callback) -> ++ sessionData.revisions[inception] 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.declarations.pop() - sessionData.doneProvisions &= (1 << sessionData.provisions.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) - + sessionData.doneProvisions &= (2 << sessionData.virtualParameters.length) - 1 rev = sessionData.revisions[sessionData.revisions.length - 1] sessionData.deviceData.timestamps.collapse(rev + 1) sessionData.deviceData.attributes.collapse(rev + 1) @@ -1527,7 +1495,8 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> operation = { type: 'Download' timestamp: sessionData.timestamp - provisions: {} + provisions: sessionData.provisions + channels: sessionData.channels retries: {} args: { instance: rpcReq.instance @@ -1537,12 +1506,9 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> } } - for provision, i in sessionData.provisions[0] - channel = sessionData.channels[i] - operation.provisions[channel] ?= [] - operation.provisions[channel].push(provision) - if sessionData.faults[channel]? - operation.retries[channel] = sessionData.faults[channel].retries or 0 + for channel of sessionData.channels + if sessionData.retries[channel]? + operation.retries[channel] = sessionData.retries[channel] sessionData.operations[rpcReq.commandKey] = operation sessionData.operationsTouched ?= {} @@ -1558,31 +1524,15 @@ rpcFault = (sessionData, id, faultResponse, callback) -> # TODO Consider handling invalid parameter faults by automatically refreshing # relevant data model portions - if not sessionData.provisions[0]?.length - throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(flt)) + fault = { + code: "cwmp.#{faultResponse.detail.faultCode}" + message: faultResponse.detail.faultString + detail: faultResponse.detail + timestamp: sessionData.timestamp + } - faults = {} - for p, i in sessionData.provisions[0] - 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) + delete sessionData.syncState + return callback(null, fault) deserialize = (sessionDataString, callback) ->