Some fixes in tasks

lib/api-functions.coffee

@@ -171,21 +171,25 @@ connectionRequest = (deviceId, callback) ->
   )
 
 
-watchTask = (taskId, timeout, callback) ->
+watchTask = (deviceId, taskId, timeout, callback) ->
   setTimeout(() ->
-    db.tasksCollection.findOne({_id : taskId}, {'_id' : 1, 'fault' : 1}, (err, task) ->
+    db.tasksCollection.findOne({_id : taskId}, {'_id' : 1}, (err, task) ->
       return callback(err) if err
 
-      if task
-        timeout -= 500
-        if task.fault?
-          callback(null, 'fault')
-        else if timeout <= 0
-          callback(null, 'timeout')
-        else
-          watchTask(taskId, timeout, callback)
-      else
-        callback(null, 'completed')
+      if not task
+        return callback(null, 'completed')
+
+      db.faultsCollection.findOne({_id : "#{deviceId}:task_#{taskId}"}, {'_id' : 1}, (err, fault) ->
+        return callback(err) if err
+
+        if fault
+          return callback(null, 'fault')
+
+        if (timeout -= 500) <= 0
+          return callback(null, 'timeout')
+
+        watchTask(deviceId, taskId, timeout, callback)
+      )
     )
   , 500)
 
@@ -263,7 +267,12 @@ deleteDevice = (deviceId, callback) ->
         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)
+          db.redisClient.del("#{deviceId}_presets_hash",
+            "#{deviceId}_inform_hash",
+            "#{deviceId}_faults",
+            "#{deviceId}_tasks",
+            "#{deviceId}_operations",
+            callback)
         )
       )
     )

lib/cwmp.coffee

@@ -339,15 +339,6 @@ nextRpc = (currentRequest) ->
     return throwError(err, currentRequest.httpResponse) if err
 
     if fault
-      for channel of currentRequest.sessionData.channels
-        if channel.startsWith('task_')
-          taskId = channel.slice(5)
-          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)
       return nextRpc(currentRequest)
@@ -373,11 +364,20 @@ nextRpc = (currentRequest) ->
     session.clearProvisions(currentRequest.sessionData)
 
     for task in currentRequest.sessionData.tasks
+      channel = "task_#{task._id}"
+
       # Delete if expired
       if task.expiry <= currentRequest.sessionData.timestamp
         util.log("#{currentRequest.sessionData.deviceId}: Task is expired #{task.name}(#{task._id})")
         currentRequest.sessionData.doneTasks ?= []
         currentRequest.sessionData.doneTasks.push(String(task._id))
+        if channel of currentRequest.sessionData.faults
+          delete currentRequest.sessionData.faults[channel]
+          currentRequest.sessionData.faultsTouched ?= {}
+          currentRequest.sessionData.faultsTouched[channel] = true
+        continue
+
+      if currentRequest.sessionData.faults[channel]
         continue
 
       switch task.name
@@ -426,7 +426,7 @@ sendRpcRequest = (currentRequest, id, rpcRequest) ->
       if isNew
         util.log("#{currentRequest.sessionData.deviceId}: New device registered")
 
-      db.clearTasks(currentRequest.sessionData.doneTasks, (err) ->
+      db.clearTasks(currentRequest.sessionData.deviceId, currentRequest.sessionData.doneTasks, (err) ->
         return throwError(err, currentRequest.httpResponse) if err
 
         counter = 3
@@ -651,15 +651,6 @@ listener = (httpRequest, httpResponse) ->
           return throwError(err, currentRequest.httpResponse) if err
 
           if fault
-            for channel of currentRequest.sessionData.channels
-              if channel.startsWith('task_')
-                taskId = channel.slice(5)
-                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)

lib/db.coffee

@@ -451,10 +451,10 @@ saveDevice = (deviceId, deviceData, isNew, callback) ->
 
 
 getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) ->
-  redisClient.mget("#{deviceId}_no_tasks", "#{deviceId}_faults", "#{deviceId}_operations", (err, res) ->
+  redisClient.mget("#{deviceId}_tasks", "#{deviceId}_faults", "#{deviceId}_operations", (err, res) ->
     return callback(err) if err
 
-    if res[0]?
+    if res[0]
       tasks = JSON.parse(res[0])
 
     if res[1]?
@@ -490,7 +490,7 @@ getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) ->
         else
           exp = CACHE_DURATION
 
-        redisClient.setex("#{deviceId}_no_tasks", exp, 1, (err) ->
+        redisClient.setex("#{deviceId}_tasks", exp, JSON.stringify(dueTasks), (err) ->
           if err
             callback?(err)
             return callback = null
@@ -526,7 +526,6 @@ getFaults = (deviceId, callback) ->
       delete r.channel
       delete r.device
       r.timestamp = +r.timestamp
-      r.expiry = +r.expiry if r.expiry?
       r.provisions = JSON.parse(r.provisions)
       faults[channel] = r
 
@@ -559,10 +558,7 @@ deleteFault = (deviceId, channel, callback) ->
 
 
 getDueTasks = (deviceId, timestamp, callback) ->
-  cur = tasksCollection.find({
-    'device' : deviceId,
-    'fault' : {'$exists': false}
-    }).sort(['timestamp'])
+  cur = tasksCollection.find({'device' : deviceId}).sort(['timestamp'])
 
   tasks = []
 
@@ -573,6 +569,7 @@ getDueTasks = (deviceId, timestamp, callback) ->
       return callback(null, tasks, null)
 
     task.timestamp = +task.timestamp if task.timestamp?
+    task.expiry = +task.expiry if task.expiry?
 
     if task.timestamp >= timestamp
       return callback(null, tasks, +task.timestamp)
@@ -599,11 +596,14 @@ getDueTasks = (deviceId, timestamp, callback) ->
   )
 
 
-clearTasks = (taskIds, callback) ->
+clearTasks = (deviceId, taskIds, callback) ->
   if not taskIds?.length
     return callback()
 
-  tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, callback)
+  tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, (err) ->
+    return callback(err) if err
+    redisClient.del("#{deviceId}_tasks", callback)
+  )
 
 
 getOperations = (deviceId, callback) ->

lib/nbi.coffee

@@ -235,17 +235,20 @@ listener = (request, response) ->
         faultId = querystring.unescape(FAULTS_REGEX.exec(urlParts.pathname)[1])
         deviceId = faultId.split(':', 1)[0]
         channel = faultId.slice(deviceId.length + 1)
-        db.redisClient.del("#{deviceId}_faults", (err) ->
-          return throwError(err, response) if err
         db.faultsCollection.remove({_id : faultId}, (err) ->
           return throwError(err, response) if err
-            if channel.startsWith('_task_')
-              return db.tasksCollection.remove({_id : new mongodb.ObjectID(channel.slice(6))}, (err) ->
+          if channel.startsWith('task_')
+            return db.tasksCollection.remove({_id : new mongodb.ObjectID(channel.slice(5))}, (err) ->
+              return throwError(err, response) if err
+              db.redisClient.del("#{deviceId}_tasks", "#{deviceId}_faults", (err) ->
                 return throwError(err, response) if err
                 response.writeHead(200)
                 response.end()
               )
+            )
 
+          db.redisClient.del("#{deviceId}_faults", (err) ->
+            return throwError(err, response) if err
             response.writeHead(200)
             response.end()
           )
@@ -261,7 +264,7 @@ listener = (request, response) ->
           task.device = deviceId
           apiFunctions.insertTasks(task, (err) ->
             return throwError(err, response) if err
-            db.redisClient.del("#{deviceId}_no_tasks", (err) ->
+            db.redisClient.del("#{deviceId}_tasks", (err) ->
               return throwError(err, response) if err
 
               if urlParts.query.connection_request?
@@ -270,7 +273,7 @@ listener = (request, response) ->
                     response.writeHead(202, err.message, {'Content-Type' : 'application/json'})
                     response.end(JSON.stringify(task))
                   else
-                    apiFunctions.watchTask(task._id, config.get('DEVICE_ONLINE_THRESHOLD', deviceId), (err, status) ->
+                    apiFunctions.watchTask(deviceId, task._id, config.get('DEVICE_ONLINE_THRESHOLD', deviceId), (err, status) ->
                       return throwError(err, response) if err
 
                       if status is 'timeout'
@@ -320,13 +323,16 @@ listener = (request, response) ->
             deviceId = task.device
             db.tasksCollection.remove({'_id' : taskId}, (err) ->
               return throwError(err, response) if err
-              db.faultsCollection.remove({_id : "#{deviceId}:_task_#{String(taskId)}"}, (err) ->
+              db.faultsCollection.remove({_id : "#{deviceId}:task_#{String(taskId)}"}, (err) ->
+                return throwError(err, response) if err
+                db.redisClient.del("#{deviceId}_tasks", "#{deviceId}_faults", (err) ->
                   return throwError(err, response) if err
                   response.writeHead(200)
                   response.end()
                 )
               )
             )
+          )
         else
           response.writeHead 405, {'Allow': 'PUT DELETE'}
           response.end('405 Method Not Allowed')
@@ -335,18 +341,15 @@ listener = (request, response) ->
           db.tasksCollection.findOne({'_id' : taskId}, {'device' : 1}, (err, task) ->
             return throwError(err, response) if err
             deviceId = task.device
-            db.tasksCollection.update({_id : taskId}, {$unset : {fault : 1}, $set : {timestamp : new Date()}}, (err, count) ->
+            db.faultsCollection.remove({_id : "#{deviceId}:task_#{String(taskId)}"}, (err) ->
               return throwError(err, response) if err
-              db.faultsCollection.remove({_id : "#{deviceId}:_task_#{String(taskId)}"}, (err) ->
-                return throwError(err, response) if err
-                db.redisClient.del("#{deviceId}_no_tasks", "#{deviceId}_faults", (err) ->
+              db.redisClient.del("#{deviceId}_faults", (err) ->
                 return throwError(err, response) if err
                 response.writeHead(200)
                 response.end()
               )
             )
           )
-          )
         else
           response.writeHead 405, {'Allow': 'POST'}
           response.end('405 Method Not Allowed')