From 3d221afc279347967a7aec4e809bcdec02738760 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Wed, 30 Nov 2016 16:27:01 -0800 Subject: [PATCH] Implement support for download RPC This uses special parameters (Downloads.) to trigger file download making it poossible to use from scripts declaratively. Resolves: #171 --- lib/api-functions.coffee | 5 +- lib/cache.coffee | 37 +++- lib/config.coffee | 1 + lib/cwmp.coffee | 120 ++++++++----- lib/db.coffee | 111 ++++++++++-- lib/instance-set.js | 8 + lib/session.coffee | 365 +++++++++++++++++++++++++++++++++++++-- lib/soap.coffee | 20 ++- 8 files changed, 584 insertions(+), 83 deletions(-) diff --git a/lib/api-functions.coffee b/lib/api-functions.coffee index 1265645..6432fef 100644 --- a/lib/api-functions.coffee +++ b/lib/api-functions.coffee @@ -218,7 +218,10 @@ deleteDevice = (deviceId, callback) -> return callback(err) if err db.faultsCollection.remove({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}, (err) -> return callback(err) if err - db.redisClient.del("#{deviceId}_presets_hash", "#{deviceId}_inform_hash", "#{deviceId}_faults", "#{deviceId}_no_tasks", callback) + db.operationsCollection.remove({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}, (err) -> + return callback(err) if err + db.redisClient.del("#{deviceId}_presets_hash", "#{deviceId}_inform_hash", "#{deviceId}_faults", "#{deviceId}_no_tasks", callback) + ) ) ) ) diff --git a/lib/cache.coffee b/lib/cache.coffee index 905c0a3..2f4a78c 100644 --- a/lib/cache.coffee +++ b/lib/cache.coffee @@ -31,6 +31,7 @@ hash = null presets = null provisions = null virtualParameters = null +files = null UNLOCK_SCRIPT = 'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end' @@ -91,7 +92,7 @@ refresh = (callback) -> return callback() lock('presets_hash_lock', 3000, (err, unlockOrExtend) -> - counter = 3 + counter = 4 db.presetsCollection.find().toArray((err, res) -> if err callback(err) if -- counter >= 0 @@ -198,11 +199,33 @@ refresh = (callback) -> return callback() ) ) + + db.filesCollection.find().toArray((err, res) -> + if err + callback(err) if -- counter >= 0 + counter = 0 + return + + files = {} + for r in res + id = r.filename or r._id.toString() + files[id] = {} + files[id].length = r.length + files[id].md5 = r.md5 + files[id].contentType = r.contentType + + if -- counter == 0 + computeHash() + db.redisClient.setex("presets_hash", 300, hash, (err) -> + unlockOrExtend(0) + nextRefresh = now + REFRESH + return callback() + ) + ) ) ) - getPresets = (callback) -> if Date.now() < nextRefresh return callback(null, hash, presets) @@ -230,6 +253,15 @@ getVirtualParameters = (callback) -> ) +getFiles = (callback) -> + if Date.now() < nextRefresh + return callback(null, hash, files) + + refresh((err) -> + return callback(err, hash, files) + ) + + getProvisionsAndVirtualParameters = (callback) -> if Date.now() < nextRefresh return callback(null, hash, provisions, virtualParameters) @@ -240,4 +272,5 @@ getProvisionsAndVirtualParameters = (callback) -> exports.getPresets = getPresets exports.getProvisions = getProvisions +exports.getFiles = getFiles exports.getProvisionsAndVirtualParameters = getProvisionsAndVirtualParameters diff --git a/lib/config.coffee b/lib/config.coffee index d2111b2..81d0400 100644 --- a/lib/config.coffee +++ b/lib/config.coffee @@ -41,6 +41,7 @@ options = { FS_SSL : {type : 'bool', default : false}, FS_IP : {type : 'string', default : '192.168.0.1'}, + DOWNLOAD_TIMEOUT: {type : 'int', default : 3600}, PRESETS_CACHE_DURATION : {type : 'int', default : 86400}, LOG_INFORMS : {type : 'bool', default : true}, DEBUG : {type : 'bool', default : false}, diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index e1e453a..21d436e 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -104,6 +104,26 @@ inform = (currentRequest, cwmpRequest) -> ) +transferComplete = (currentRequest, cwmpRequest) -> + session.transferComplete(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse, faults) -> + throw err if err + + 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})") + + res = soap.response({ + id : cwmpRequest.id, + methodResponse : rpcResponse, + cwmpVersion : currentRequest.sessionData.cwmpVersion + }) + writeResponse(currentRequest, res) + ) + + testSchedule = (schedule, timestamp) -> range = schedule.schedule.nextRange(10, new Date(timestamp)) prev = schedule.schedule.prevRange(1, new Date(timestamp), new Date(timestamp - schedule.duration)) @@ -207,8 +227,12 @@ applyPresets = (currentRequest) -> if query.testFilter(parameters, p.precondition) session.addProvisions(currentRequest.sessionData, p.channel, p.provisions) - session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest) -> + session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest, doneProvisions) -> throw err if err + + for p, i in doneProvisions or [] + delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]] + sendRpcRequest(currentRequest, id, rpcRequest) ) ) @@ -222,8 +246,7 @@ nextRpc = (currentRequest) -> return sendRpcRequest(currentRequest, id, rpcRequest) for p, i in doneProvisions or [] - if currentRequest.sessionData.faults? - delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]] + delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]] if p[0] == '_task' currentRequest.sessionData.doneTasks ?= [] @@ -257,6 +280,8 @@ nextRpc = (currentRequest) -> session.addProvisions(currentRequest.sessionData, "_task_#{task._id}", [['_task', task._id, 'reboot']]) when 'factoryReset' session.addProvisions(currentRequest.sessionData, "_task_#{task._id}", [['_task', task._id, 'factoryReset']]) + when 'download' + session.addProvisions(currentRequest.sessionData, "_task_#{task._id}", [['_task', task._id, 'download', task.fileType, task.fileName, task.targetFileName]]) else throw new Error('Task name not recognized') @@ -289,6 +314,28 @@ sendRpcRequest = (currentRequest, id, rpcRequest) -> ) return + if rpcRequest.type is 'Download' + if not rpcRequest.url? + FS_PORT = config.get('FS_PORT') + FS_IP = config.get('FS_IP') + FS_SSL = config.get('FS_SSL') + rpcRequest.url = if FS_SSL then 'https://' else 'http://' + rpcRequest.url += FS_IP + rpcRequest.url += ":#{FS_PORT}" if FS_PORT != 80 + rpcRequest.url += "/#{encodeURIComponent(rpcRequest.fileName)}" + + if not rpcRequest.fileSize? + return cache.getFiles((err, hash, files) -> + throw err if err + + if rpcRequest.fileName of files + rpcRequest.fileSize = files[rpcRequest.fileName].length + else + rpcRequest.fileSize = 0 + + return sendRpcRequest(currentRequest, id, rpcRequest) + ) + util.log("#{currentRequest.sessionData.deviceId}: #{rpcRequest.type} (#{id})") res = soap.response({ @@ -402,10 +449,11 @@ listener = (httpRequest, httpResponse) -> httpRequest.connection.setTimeout(sessionData.timeout * 1000) - db.getDueTasksAndFaults(deviceId, sessionData.timestamp, (err, dueTasks, faults) -> + db.getDueTasksAndFaultsAndOperations(deviceId, sessionData.timestamp, (err, dueTasks, faults, operations) -> throw err if err sessionData.tasks = dueTasks - sessionData.faults = faults if Object.keys(faults).length + sessionData.faults = faults + sessionData.operations = operations # Delete expired faults for k, v of sessionData.faults @@ -430,6 +478,8 @@ listener = (httpRequest, httpResponse) -> if cwmpRequest.methodRequest? if cwmpRequest.methodRequest.type is 'Inform' inform(currentRequest, cwmpRequest) + else if cwmpRequest.methodRequest.type is 'TransferComplete' + transferComplete(currentRequest, cwmpRequest) else if cwmpRequest.methodRequest.type is 'GetRPCMethods' util.log("#{currentRequest.sessionData.deviceId}: GetRPCMethods") res = soap.response({ @@ -446,44 +496,17 @@ listener = (httpRequest, httpResponse) -> nextRpc(currentRequest) ) else if cwmpRequest.fault? - session.rpcFault(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.fault, (err, flt) -> + session.rpcFault(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.fault, (err, faults) -> throw err if err - if not flt? - return nextRpc(currentRequest) - - if not currentRequest.sessionData.provisions.length - throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(flt)) - - faults = {} - currentRequest.sessionData.faults ?= {} - - for p, i in currentRequest.sessionData.provisions[0] - channel = currentRequest.sessionData.channels[i] - fault = faults[channel] - if not fault? - fault = { - device: currentRequest.sessionData.deviceId - channel: channel - timestamp: currentRequest.sessionData.timestamp - provisions: [] - fault: flt - } - - if p[0] == '_task' - for t, j in currentRequest.sessionData.tasks - if t._id == p[1] - currentRequest.sessionData.tasks.splice(j, 1) - fault.expiry = t.expiry if t.expiry? - break - - faults[channel] = fault - - fault.provisions.push(p) - for channel, fault of faults - if currentRequest.sessionData.faults[channel]? - fault.retries = (currentRequest.sessionData.faults[channel].retries ? 0) + 1 + 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 @@ -491,12 +514,21 @@ listener = (httpRequest, httpResponse) -> currentRequest.sessionData.faults[channel] = fault util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})") - session.clearProvisions(currentRequest.sessionData) nextRpc(currentRequest) ) - else - # cpe sent empty response. add presets - nextRpc(currentRequest) + else # CPE sent empty response + session.timeoutOperations(currentRequest.sessionData, (err, faults) -> + throw err if err + + 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})") + + nextRpc(currentRequest) + ) ) ) diff --git a/lib/db.coffee b/lib/db.coffee index 5afe7fe..7dbb06e 100644 --- a/lib/db.coffee +++ b/lib/db.coffee @@ -28,10 +28,12 @@ objectsCollection = null provisionsCollection = null virtualParametersCollection = null faultsCollection = null +filesCollection = null +operationsCollection = null connect = (callback) -> - callbackCounter = 9 + callbackCounter = 10 mongodb.MongoClient.connect(config.get('MONGODB_CONNECTION_URL'), {db:{w:1},server:{autoReconnect:true}}, (err, db) -> return callback(err) if err exports.mongoDb = db @@ -94,6 +96,13 @@ connect = (callback) -> return callback(err) ) + db.collection('operations', (err, collection) -> + exports.operationsCollection = operationsCollection = collection + if --callbackCounter == 0 or err + callbackCounter = 0 + return callback(err) + ) + exports.redisClient = redisClient = redis.createClient(config.get('REDIS_PORT'), config.get('REDIS_HOST')) redisClient.select(config.get('REDIS_DB'), (err) -> if --callbackCounter == 0 or err @@ -441,18 +450,21 @@ saveDevice = (deviceId, deviceData, isNew, callback) -> ) -getDueTasksAndFaults = (deviceId, timestamp, callback) -> - redisClient.mget("#{deviceId}_no_tasks", "#{deviceId}_faults", (err, res) -> +getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) -> + redisClient.mget("#{deviceId}_no_tasks", "#{deviceId}_faults", "#{deviceId}_operations", (err, res) -> return callback(err) if err if res[0]? - tasks = [] + tasks = JSON.parse(res[0]) if res[1]? faults = JSON.parse(res[1]) - if tasks? and faults? - return callback(null, tasks, faults) + if res[2]? + operations = JSON.parse(res[2]) + + if tasks? and faults? and operations? + return callback(null, tasks, faults, operations) CACHE_DURATION = config.get('PRESETS_CACHE_DURATION', deviceId) @@ -463,8 +475,8 @@ getDueTasksAndFaults = (deviceId, timestamp, callback) -> return callback = null faults = flts - if tasks? - return callback(null, tasks, faults) + if tasks? and operations? + return callback(null, tasks, faults, operations) ) if not tasks? @@ -484,10 +496,22 @@ getDueTasksAndFaults = (deviceId, timestamp, callback) -> return callback = null tasks = dueTasks - if faults? - return callback(null, tasks, faults) + if faults? and operations? + return callback(null, tasks, faults, operations) ) ) + + if not operations? + getOperations(deviceId, (err, _operations) -> + if err + callback?(err) + return callback = null + + operations = _operations + if tasks? and faults? + return callback(null, tasks, faults, operations) + ) + ) @@ -499,6 +523,8 @@ getFaults = (deviceId, callback) -> for r in res channel = r._id.slice(deviceId.length + 1) delete r._id + delete r.channel + delete r.device r.timestamp = +r.timestamp r.expiry = +r.expiry if r.expiry? faults[channel] = r @@ -533,7 +559,23 @@ getDueTasks = (deviceId, timestamp, callback) -> task._id = String(task._id) tasks.push(task) - cur.nextObject(f) + + # For API compatibility + if task.name is 'download' and task.file? + if mongodb.ObjectID.isValid(task.file) + q = {_id: {'$in' : [task.file, new mongodb.ObjectID(task.file)]}} + else + q = {_id: task.file} + + filesCollection.find(q).toArray((err, res) -> + return callback(err) if err + if res[0]? + task.fileType ?= res[0].metadata.fileType + task.fileName ?= res[0].filename or res[0]._id.toString() + cur.nextObject(f) + ) + else + cur.nextObject(f) ) @@ -570,6 +612,8 @@ syncFaults = (deviceId, faults, callback) -> 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? @@ -597,10 +641,53 @@ syncFaults = (deviceId, faults, callback) -> ) +getOperations = (deviceId, callback) -> + operationsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) -> + return callback(err) if err + + operations = {} + for r in res + commandKey = r._id.slice(deviceId.length + 1) + delete r._id + r.timestamp = +r.timestamp + r.args = JSON.parse(r.args) if r.args + r.provisions = JSON.parse(r.provisions) + r.retries = JSON.parse(r.retries) + operations[commandKey] = r + + CACHE_DURATION = config.get('PRESETS_CACHE_DURATION', deviceId) + + redisClient.setex("#{deviceId}_operations", CACHE_DURATION, JSON.stringify(operations), (err) -> + callback(err, operations) + ) + ) + + +saveOperation = (deviceId, commandKey, operation, callback) -> + operation._id = "#{deviceId}:#{commandKey}" + operation.timestamp = new Date(operation.timestamp) + operation.provisions = JSON.stringify(operation.provisions) + operation.retries = JSON.stringify(operation.retries) + operation.args = JSON.stringify(operation.args) + operationsCollection.save(operation, (err) -> + return callback(err) if err + redisClient.del("#{deviceId}_operations", callback) + ) + + +deleteOperation = (deviceId, commandKey, callback) -> + operationsCollection.remove({_id: "#{deviceId}:#{commandKey}"}, (err) -> + return callback(err) if err + redisClient.del("#{deviceId}_operations", callback) + ) + + exports.connect = connect exports.disconnect = disconnect exports.fetchDevice = fetchDevice exports.saveDevice = saveDevice -exports.getDueTasksAndFaults = getDueTasksAndFaults +exports.getDueTasksAndFaultsAndOperations = getDueTasksAndFaultsAndOperations exports.clearTasks = clearTasks exports.syncFaults = syncFaults +exports.saveOperation = saveOperation +exports.deleteOperation = deleteOperation diff --git a/lib/instance-set.js b/lib/instance-set.js index 953c5a5..682aac8 100644 --- a/lib/instance-set.js +++ b/lib/instance-set.js @@ -123,6 +123,14 @@ class InstanceSet { values() { return this.set.values(); } + + clear() { + this.set.clear(); + } + + get size() { + return this.set.size; + } } module.exports = InstanceSet; diff --git a/lib/session.coffee b/lib/session.coffee index 6f627c8..7dc2165 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -23,7 +23,7 @@ sandbox = require './sandbox' cache = require './cache' extensions = require './extensions' PathSet = require './path-set' -VersionedMap = require './versioned-map.js' +VersionedMap = require './versioned-map' InstanceSet = require './instance-set' @@ -156,12 +156,139 @@ inform = (sessionData, rpcReq, callback) -> ) +transferComplete = (sessionData, rpcReq, callback) -> + commandKey = rpcReq.commandKey + operation = sessionData.operations[commandKey] + if not operation? + # TODO Show a warning instead + throw new Error('Invalid command key') + instance = operation.args.instance + faults = {} + + delete sessionData.operations[commandKey] + sessionData.operationsTouched ?= {} + sessionData.operationsTouched[commandKey] = 1 + + 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 + } + + if operation.retries[channel]? + faults[channel].retries = operation.retries[channel] + 1 + + return callback(err, {type : 'TransferCompleteResponse'}, faults) + ) + + loadPath(sessionData, ['Downloads', instance, '*']) + + loadParameters(sessionData, (err) -> + return callback(err) if err + + toClear = null + timestamp = sessionData.timestamp + sessionData.iteration + 1 + + p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastDownload']) + toClear = device.set(sessionData.deviceData, p, timestamp, + {value: [timestamp, [operation.timestamp, 'xsd:dateTime']]}, toClear) + + p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastFileType']) + toClear = device.set(sessionData.deviceData, p, timestamp, + {value: [timestamp, [operation.args.fileType, 'xsd:string']]}, toClear) + + p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastFileName']) + toClear = device.set(sessionData.deviceData, p, timestamp, + {value: [timestamp, [operation.args.fileName, 'xsd:string']]}, toClear) + + p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastTargetFileName']) + toClear = device.set(sessionData.deviceData, p, timestamp, + {value: [timestamp, [operation.args.targetFileName, 'xsd:string']]}, toClear) + + p = sessionData.deviceData.paths.add(['Downloads', instance, 'StartTime']) + toClear = device.set(sessionData.deviceData, p, timestamp, + {value: [timestamp, [+rpcReq.startTime, 'xsd:dateTime']]}, toClear) + + p = sessionData.deviceData.paths.add(['Downloads', instance, 'CompleteTime']) + toClear = device.set(sessionData.deviceData, p, timestamp, + {value: [timestamp, [+rpcReq.completeTime, 'xsd:dateTime']]}, toClear) + + clear(sessionData, toClear, (err) -> + return callback(err, {type : 'TransferCompleteResponse'}, faults) + ) + ) + + +revertDownloadParameters = (sessionData, instance, callback) -> + loadPath(sessionData, ['Downloads', instance, '*']) + + loadParameters(sessionData, (err) -> + return callback(err) if err + + timestamp = sessionData.timestamp + sessionData.iteration + 1 + + p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastDownload']) + LastDownload = sessionData.deviceData.attributes.get(p) + + p = sessionData.deviceData.paths.add(['Downloads', instance, 'Download']) + toClear = device.set(sessionData.deviceData, p, timestamp, + {value: [timestamp, [LastDownload?.value[1]?[0] or 0, 'xsd:dateTime']]}, toClear) + + clear(sessionData, toClear, callback) + ) + + +timeoutOperations = (sessionData, callback) -> + faults = {} + counter = 1 + + for commandKey, operation of sessionData.operations + if operation.type isnt 'Download' + return callback(new Error("Unknown operation type #{operation.type}")) + + DOWNLOAD_TIMEOUT = config.get('DOWNLOAD_TIMEOUT', sessionData.deviceId) * 1000 + + if sessionData.timestamp > operation.timestamp + DOWNLOAD_TIMEOUT + delete sessionData.operations[commandKey] + 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' + } + + if operation.retries[channel]? + faults[channel].retries = operation.retries[channel] + 1 + + ++ counter + revertDownloadParameters(sessionData, operation.args.instance, (err) -> + -- counter + if err and counter > 0 + counter = 0 + + return callback(err, faults) if counter == 0 + ) + + -- counter + return callback(null, faults) 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] ?= [] @@ -186,6 +313,7 @@ clearProvisions = (sessionData) -> sessionData.deviceData.attributes.collapse(1) delete sessionData.syncState + delete sessionData.rpcRequest sessionData.provisions = [] sessionData.channels = [] sessionData.declarations = [] @@ -239,6 +367,12 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback) allDeclarations.push([['Reboot'], 1, {value: 1}, null, {value: [sessionData.timestamp]}]) when 'factoryReset' allDeclarations.push([['FactoryReset'], 1, {value: 1}, null, {value: [sessionData.timestamp]}]) + when 'download' + alias = "[FileType:#{JSON.stringify(provision[3] or '')},FileName:#{JSON.stringify(provision[4] or '')},TargetFileName:#{JSON.stringify(provision[5] or '')}]" + allDeclarations.push([common.parsePath("Downloads.#{alias}"), + 1, {}, 1, {}]) + allDeclarations.push([common.parsePath("Downloads.#{alias}.Download"), + 1, {value: 1}, null, {value: [sessionData.timestamp]}]) continue ++ counter @@ -339,6 +473,11 @@ runDeclarations = (sessionData, declarations) -> virtualParameterDeclarations: [] instancesToDelete: new Map() instancesToCreate: new Map() + + downloadsToDelete: new Set() + downloadsToCreate: new InstanceSet() + downloadsValues: new Map() + downloadsDownload: new Map() } allDeclareTimestamps = new Map() @@ -361,6 +500,14 @@ runDeclarations = (sessionData, declarations) -> if (path.alias | path.wildcard) & 1 or path[0] == 'FactoryReset' sessionData.deviceData.paths.add(['FactoryReset']) + if path[0] == 'Files' and + !(path.alias | path.wildcard) and + path.length == 3 and + path[2] == 'Download' and + (declaration[4]?.value? or sessionData.deviceData.attributes.has(path)) + sessionData.syncState.download ?= new Map() + sessionData.syncState.download.set(path, 0) + if path.alias aliasDecs = device.getAliasDeclarations(path, declaration[1] or 1) for ad in aliasDecs @@ -527,11 +674,12 @@ rpcRequest = (sessionData, _declarations, callback) -> device.clearTrackers(sessionData.deviceData, 'prerequisite') return rpcRequest(sessionData, null, callback) - # Update tags toClear = null + timestamp = sessionData.timestamp + sessionData.iteration + 1 + + # Update tags sessionData.syncState.tags.forEach((v, p) -> c = sessionData.deviceData.attributes.get(p) - timestamp = sessionData.timestamp + sessionData.iteration + 1 if v and not c? toClear = device.set(sessionData.deviceData, p, timestamp, {object: [timestamp, false], writable: [timestamp, true], value: [timestamp, [true, 'xsd:boolean']]}, toClear) else if c? and not v @@ -546,6 +694,65 @@ rpcRequest = (sessionData, _declarations, callback) -> toClear = device.set(sessionData.deviceData, ['Tags'], timestamp, null, toClear) ) + # Downloads + index = null + sessionData.syncState.downloadsToCreate.forEach((instance) -> + if not index? + index = 0 + iter = sessionData.deviceData.paths.subset(['Downloads', '*']) + while p = iter.next().value + if +p[1] > index and sessionData.deviceData.attributes.has(p) + index = +p[1] + + ++ index + + toClear = device.set(sessionData.deviceData, ['Downloads'], + timestamp, + {object: [timestamp, 1], writable: [timestamp, 1]}, toClear) + + toClear = device.set(sessionData.deviceData, ['Downloads', "#{index}"], + timestamp, + {object: [timestamp, 1], writable: [timestamp, 1]}, toClear) + + params = { + 'FileType': {writable: 1, value: [instance.FileType or '', 'xsd:string']} + 'FileName': {writable: 1, value: [instance.FileName or '', 'xsd:string']} + 'TargetFileName': {writable: 1, value: [instance.TargetFileName or '', 'xsd:string']} + 'Download': {writable: 1, value: [instance.Download or 0, 'xsd:dateTime']} + 'LastFileType': {writable: 0, value: ['', 'xsd:string']} + 'LastFileName': {writable: 0, value: ['', 'xsd:string']} + 'LastTargetFileName': {writable: 0, value: ['', 'xsd:string']} + 'LastDownload': {writable: 0, value: [0, 'xsd:dateTime']} + 'StartTime': {writable: 0, value: [0, 'xsd:dateTime']} + 'CompleteTime': {writable: 0, value: [0, 'xsd:dateTime']} + } + + for k, v of params + toClear = device.set(sessionData.deviceData, ['Downloads', "#{index}", k], + timestamp, {object: [timestamp, 0], writable: [timestamp, v.writable], value: [timestamp, v.value]}, toClear) + + toClear = device.set(sessionData.deviceData, ['Downloads', "#{index}", '*'], + timestamp, null, toClear) + ) + sessionData.syncState.downloadsToCreate.clear() + + sessionData.syncState.downloadsToDelete.forEach((instance) -> + toClear = device.set(sessionData.deviceData, instance, timestamp, null, toClear) + sessionData.syncState.downloadsValues.forEach((v, p) -> + if p[1] == instance[1] + sessionData.syncState.downloadsValues.delete(p) + ) + ) + sessionData.syncState.downloadsToDelete.clear() + + sessionData.syncState.downloadsValues.forEach((v, p) -> + if attrs = sessionData.deviceData.attributes.get(p) + if attrs.writable?[1] and attrs.value? + v = device.sanitizeParameterValue([v, attrs.value[1][1]]) + if v[0] != attrs.value[1][0] + toClear = device.set(sessionData.deviceData, p, timestamp, {value: [timestamp, v]}, toClear) + ) + if toClear return clear(sessionData, toClear, (err) -> return callback(err) if err @@ -747,6 +954,8 @@ generateSetRpcRequest = (sessionData) -> if not (syncState = sessionData.syncState)? return + deviceData = sessionData.deviceData + # Delete instance iter = syncState.instancesToDelete.values() while instances = iter.next().value @@ -812,6 +1021,22 @@ generateSetRpcRequest = (sessionData) -> type: 'FactoryReset' } + # Download + iter = syncState.downloadsDownload.entries() + while pair = iter.next().value + if not (pair[1] <= deviceData.attributes.get(pair[0])?.value?[1][0]) + fileTypePath = deviceData.paths.subset(pair[0].slice(0, -1).concat('FileType')).next().value + fileNamePath = deviceData.paths.subset(pair[0].slice(0, -1).concat('FileName')).next().value + targetFileNamePath = deviceData.paths.subset(pair[0].slice(0, -1).concat('TargetFileName')).next().value + return { + type: 'Download' + commandKey: generateRpcId(sessionData) + instance: pair[0][1] + fileType: deviceData.attributes.get(fileTypePath)?.value?[1][0] + fileName: deviceData.attributes.get(fileNamePath)?.value?[1][0] + targetFileName: deviceData.attributes.get(targetFileNamePath)?.value?[1][0] + } + return null @@ -905,6 +1130,12 @@ processDeclarations = (sessionData, allDeclareTimestamps, allDeclareAttributeTim syncState.tags.set(currentPath, device.sanitizeParameterValue([declareAttributeValues.value[0], 'xsd:boolean'])[0]) when 'Events', 'DeviceID' then # Do nothing + when 'Downloads' + if currentPath.length == 3 and currentPath.wildcard == 0 and declareAttributeValues.value? + if currentPath[2] == 'Download' + syncState.downloadsDownload.set(currentPath, declareAttributeValues.value[0]) + else + syncState.downloadsValues.set(currentPath, declareAttributeValues.value[0]) when 'VirtualParameters' if currentPath.length <= 2 d = null @@ -999,15 +1230,20 @@ loadPath = (sessionData, path, depth) -> processInstances = (sessionData, parent, parameters, keys, minInstances, maxInstances) -> - instancesToDelete = sessionData.syncState.instancesToDelete.get(parent) - if not instancesToDelete? - instancesToDelete = new Set() - sessionData.syncState.instancesToDelete.set(parent, instancesToDelete) + if parent[0] == 'Downloads' + return if parent.length != 1 + instancesToDelete = sessionData.syncState.downloadsToDelete + instancesToCreate = sessionData.syncState.downloadsToCreate + else + instancesToDelete = sessionData.syncState.instancesToDelete.get(parent) + if not instancesToDelete? + instancesToDelete = new Set() + sessionData.syncState.instancesToDelete.set(parent, instancesToDelete) - instancesToCreate = sessionData.syncState.instancesToCreate.get(parent) - if not instancesToCreate? - instancesToCreate = new InstanceSet() - sessionData.syncState.instancesToCreate.set(parent, instancesToCreate) + instancesToCreate = sessionData.syncState.instancesToCreate.get(parent) + if not instancesToCreate? + instancesToCreate = new InstanceSet() + sessionData.syncState.instancesToCreate.set(parent, instancesToCreate) counter = 0 for p in parameters @@ -1223,6 +1459,49 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> toClear = device.set(sessionData.deviceData, common.parsePath('FactoryReset'), timestamp + 1, {value: [timestamp + 1, [timestamp + 1, 'xsd:dateTime']]}, toClear) + when 'DownloadResponse' + toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'Download'], + timestamp + 1, {value: [timestamp + 1, [timestamp + 1, 'xsd:dateTime']]}, toClear) + + if rpcRes.status == 0 + toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'LastDownload'], + timestamp + 1, {value: [timestamp + 1, [timestamp + 1, 'xsd:dateTime']]}, toClear) + toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'LastFileType'], + timestamp + 1, {value: [timestamp + 1, [rpcReq.fileType, 'xsd:dateTime']]}, toClear) + toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'LastFileName'], + timestamp + 1, {value: [timestamp + 1, [rpcReq.fileType, 'xsd:dateTime']]}, toClear) + toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'LastTargetFileName'], + timestamp + 1, {value: [timestamp + 1, [rpcReq.fileType, 'xsd:dateTime']]}, toClear) + + toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'StartTime'], + timestamp + 1, {value: [timestamp + 1, [+rpcRes.startTime, 'xsd:dateTime']]}, toClear) + toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'CompleteTime'], + timestamp + 1, {value: [timestamp + 1, [+rpcRes.completeTime, 'xsd:dateTime']]}, toClear) + else + operation = { + type: 'Download' + timestamp: sessionData.timestamp + provisions: {} + retries: {} + args: { + instance: rpcReq.instance + fileType: rpcReq.fileType + fileName: rpcReq.fileName + targetFileName: rpcReq.targetFileName + } + } + + 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 + + sessionData.operations[rpcReq.commandKey] = operation + sessionData.operationsTouched ?= {} + sessionData.operationsTouched[rpcReq.commandKey] = 1 + else return callback(new Error('Response type not recognized')) @@ -1232,9 +1511,32 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> rpcFault = (sessionData, id, faultResponse, callback) -> # TODO Consider handling invalid parameter faults by automatically refreshing # relevant data model portions - delete sessionData.rpcRequest - delete sessionData.syncState - return callback(null, faultResponse) + + if not sessionData.provisions[0]?.length + throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(flt)) + + 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.Fault['FaultCode']}" + message: faultResponse.detail.Fault['FaultString'] + detail: faultResponse.detail.Fault + } + + if sessionData.faults[channel]? + fault.retries = (sessionData.faults[channel].retries or 0) + 1 + + faults[channel] = fault + + fault.provisions.push(p) + + clearProvisions(sessionData) + return callback(null, faults) deserialize = (sessionDataString, callback) -> @@ -1312,16 +1614,43 @@ serialize = (sessionData, callback) -> end = (sessionData, callback) -> + counter = 2 db.saveDevice(sessionData.deviceId, sessionData.deviceData, sessionData.new, (err) -> - return callback(err) if err - db.redisClient.del("session_#{sessionData.id}", (err) -> - callback(err, sessionData.new) - ) + -- counter + if err and counter > 0 + counter = 0 + return callback(err, sessionData.new) if counter == 0 ) + db.redisClient.del("session_#{sessionData.id}", (err) -> + -- counter + if err and counter > 0 + counter = 0 + return callback(err, sessionData.new) if counter == 0 + ) + + for k of sessionData.operationsTouched + ++ counter + if sessionData.operations[k]? + db.saveOperation(sessionData.deviceId, k, sessionData.operations[k], (err) -> + -- counter + if err and counter > 0 + counter = 0 + return callback(err, sessionData.new) if counter == 0 + ) + else + db.deleteOperation(sessionData.deviceId, k, (err) -> + -- counter + if err and counter > 0 + counter = 0 + return callback(err, sessionData.new) if counter == 0 + ) + exports.init = init +exports.timeoutOperations = timeoutOperations exports.inform = inform +exports.transferComplete = transferComplete exports.addProvisions = addProvisions exports.clearProvisions = clearProvisions exports.rpcRequest = rpcRequest diff --git a/lib/soap.coffee b/lib/soap.coffee index 035004d..6a7cff7 100644 --- a/lib/soap.coffee +++ b/lib/soap.coffee @@ -217,14 +217,22 @@ cpeDownload = (xml, methodRequest) -> el.node('Username').text(methodRequest.username ? '') el.node('Password').text(methodRequest.password ? '') el.node('FileSize').text(methodRequest.fileSize ? '0') - el.node('TargetFileName').text(methodRequest.TargetFileName ? '') + el.node('TargetFileName').text(methodRequest.targetFileName ? '') el.node('DelaySeconds').text(methodRequest.delaySeconds ? '0') el.node('SuccessURL').text(methodRequest.successUrl ? '') el.node('FailureURL').text(methodRequest.failureUrl ? '') cpeDownloadResponse = (xml) -> - {status : JSON.parse(xml.get('Status').text())} + res = { + status : parseInt(xml.get('Status').text()), + } + + if res.status == 0 + res.startTime = Date.parse(xml.get('StartTime').text()) + res.completeTime = Date.parse(xml.get('CompleteTime').text()) + + return res acsInform = (xml) -> @@ -255,8 +263,8 @@ acsTransferComplete = (xml) -> { commandKey : xml.get('CommandKey').text(), faultStruct : traverseXml(xml.get('FaultStruct')), - startTime : xml.get('StartTime').text(), # TODO convert to datetime - completeTime : xml.get('CompleteTime').text() # TODO convert to datetime + startTime : Date.parse(xml.get('StartTime').text()), + completeTime : Date.parse(xml.get('CompleteTime').text()) } @@ -367,7 +375,7 @@ exports.request = (httpRequest, cwmpVersion) -> else faultElement = xml.get('/soap-env:Envelope/soap-env:Body/soap-env:Fault', NAMESPACES[cwmpRequest.cwmpVersion]) cwmpRequest.fault = fault(faultElement) - + return cwmpRequest @@ -425,7 +433,7 @@ exports.response = (cwmpResponse) -> cpeDownload(body, cwmpResponse.methodRequest) else throw Error("Unknown method request #{cwmpResponse.methodRequest.type}") - + if env? headers['Content-Type'] = 'text/xml; charset="utf-8"' return {code: 200, headers: headers, data: new Buffer(env.doc().toString(false))}