Added options to expire tasks and prevent duplicates.

acs.coffee

@@ -136,6 +136,12 @@ runTask = (currentRequest, task, methodResponse) ->
   )
 
 
+isTaskExpired = (task) ->
+  now = Date.now()
+  if task.expires and (now - task.timestamp.getTime()) > config.DEVICE_ONLINE_THRESHOLD
+    return true
+  return false
+
 nextTask = (currentRequest) ->
   cur = db.tasksCollection.find({'device' : currentRequest.deviceId}).sort(['timestamp']).limit(1)
   cur.nextObject( (err, task) ->
@@ -171,6 +177,10 @@ nextTask = (currentRequest) ->
       # last task was faulty. Do nothing until until task is deleted
       res = tr069.response(null, cwmpResponse)
       writeResponse(currentRequest, res)
+    else if isTaskExpired(task)
+      db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) ->
+        nextTask(currentRequest)
+      )
     else
       util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})")
       runTask(currentRequest, task, {})

api.coffee

@@ -24,11 +24,8 @@ connectionRequest = (deviceId, callback) ->
     connectionRequestUrl = device.InternetGatewayDevice.ManagementServer.ConnectionRequestURL._value
     # for testing
     #connectionRequestUrl = connectionRequestUrl.replace(/^(http:\/\/)([0-9\.]+)(\:[0-9]+\/[a-zA-Z0-9]+\/?$)/, '$1192.168.1.1$3')
-    request = http.request(connectionRequestUrl, (res) ->
+    request = http.get(connectionRequestUrl, (res) ->
-      res.setEncoding('utf8')
+      callback()
-      res.on('end', () ->
-        callback()
-      )
     )
 
     request.on('error', (err) ->
@@ -43,7 +40,6 @@ connectionRequest = (deviceId, callback) ->
         request.abort()
       )
     )
-    request.end()
   )
 
 watchTask = (taskId, timeout, callback) ->
@@ -63,7 +59,7 @@ watchTask = (taskId, timeout, callback) ->
 
 sanitizeTask = (device, task) ->
   task.device = device
-  task.timestamp = mongodb.Timestamp()
+  task.timestamp = new Date()
   return task
 
 cluster = require 'cluster'
@@ -169,37 +165,34 @@ else
             # queue given task
             task = sanitizeTask(deviceId, JSON.parse(body))
 
+            if task.uniqueKey?
+              db.tasksCollection.remove({device : deviceId, uniqueKey : task.uniqueKey}, (err, removed) ->
+              )
+
             db.tasksCollection.save(task, (err) ->
               if err
                 response.writeHead(500)
                 response.end(err)
                 return
 
-              watch = () ->
-                if urlParts.query.timeout?
-                  timeout = parseInt(urlParts.query.timeout)
-                  watchTask(task._id, timeout, (err) ->
-                    if err
-                      response.writeHead(202)
-                      response.end()
-                      return
-                    response.writeHead(200)
-                    response.end()
-                  )
-                else
-                  response.writeHead(202)
-                  response.end()
-
               if urlParts.query.connection_request?
                 connectionRequest(deviceId, (err) ->
                   if err
                     response.writeHead(202)
                     response.end()
                   else
-                    watch()
+                    watchTask(task._id, config.DEVICE_ONLINE_THRESHOLD, (err) ->
+                      if err
+                        response.writeHead(202)
+                        response.end()
+                        return
+                      response.writeHead(200)
+                      response.end()
+                    )
                 )
               else
-                watch()
+                response.writeHead(202)
+                response.end()
             )
           else if urlParts.query.connection_request?
             # no task, send connection request only
@@ -236,7 +229,7 @@ else
             response.end('405 Method Not Allowed')
         else if action is '/retry'
           if request.method == 'POST'
-            db.tasksCollection.update({_id : taskId}, {$unset : {fault : 1}, $inc : {retries : 1}}, (err, count) ->
+            db.tasksCollection.update({_id : taskId}, {$unset : {fault : 1}, $inc : {retries : 1}, timestamp : new Date()}, (err, count) ->
               response.writeHead(200)
               response.end()
             )

config.coffee

@@ -16,6 +16,7 @@ exports.MEMCACHED_SOCKET = '/tmp/memcached.sock'
 exports.WORKER_RESPAWN_TIME = 3000
 exports.LOG_INFORMS = false
 exports.DEBUG_DEVICES = {} # {'202BC1-BM632w-8KA8WA1151100043' : true}
+exports.DEVICE_ONLINE_THRESHOLD = 4000
 
 parameters = require './parameters'
 aliases = {}