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

@@ -217,11 +217,14 @@ deleteDevice = (deviceId, callback) ->
     db.devicesCollection.remove({'_id' : deviceId}, (err) ->
       return callback(err) if err
       db.faultsCollection.remove({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}, (err) ->
+        return callback(err) if err
+        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)
         )
       )
     )
+  )
 
 
 exports.connectionRequest = connectionRequest

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,9 +199,31 @@ 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) ->
@@ -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

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},

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,7 +246,6 @@ 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]]
 
       if p[0] == '_task'
@@ -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,59 +496,41 @@ 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 channel, fault of faults
+            if channel.startsWith('_task_')
+              taskId = channel.slice(6)
               for t, j in currentRequest.sessionData.tasks
-                  if t._id == p[1]
+                if t._id == taskId
                   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 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})")
 
-          session.clearProvisions(currentRequest.sessionData)
           nextRpc(currentRequest)
         )
-      else
-        # cpe sent empty response. add presets
+      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)
         )
     )
+  )
 
 
 exports.listener = listener

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,6 +559,22 @@ getDueTasks = (deviceId, timestamp, callback) ->
 
     task._id = String(task._id)
     tasks.push(task)
+
+    # 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

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;

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,6 +1230,11 @@ loadPath = (sessionData, path, depth) ->
 
 
 processInstances = (sessionData, parent, parameters, keys, minInstances, maxInstances) ->
+  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()
@@ -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

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())
   }