From 310114a500193116ad285ec1d63036790481628c Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Tue, 10 Jan 2017 01:54:04 -0800 Subject: [PATCH] Some fixes in tasks --- lib/api-functions.coffee | 37 +++++++++++++++++++++++-------------- lib/cwmp.coffee | 29 ++++++++++------------------- lib/db.coffee | 20 ++++++++++---------- lib/nbi.coffee | 37 ++++++++++++++++++++----------------- 4 files changed, 63 insertions(+), 60 deletions(-) diff --git a/lib/api-functions.coffee b/lib/api-functions.coffee index 696e596..db52603 100644 --- a/lib/api-functions.coffee +++ b/lib/api-functions.coffee @@ -171,21 +171,25 @@ connectionRequest = (deviceId, callback) -> ) -watchTask = (taskId, timeout, callback) -> - setTimeout( () -> - db.tasksCollection.findOne({_id : taskId}, {'_id' : 1, 'fault' : 1}, (err, task) -> +watchTask = (deviceId, taskId, timeout, callback) -> + setTimeout(() -> + 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) ) ) ) diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index 6766d91..428dee0 100644 --- a/lib/cwmp.coffee +++ b/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) diff --git a/lib/db.coffee b/lib/db.coffee index 9cb4494..9b20060 100644 --- a/lib/db.coffee +++ b/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) -> diff --git a/lib/nbi.coffee b/lib/nbi.coffee index ef830a6..be8ac6f 100644 --- a/lib/nbi.coffee +++ b/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) -> + db.faultsCollection.remove({_id : faultId}, (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,10 +323,13 @@ 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 - response.writeHead(200) - response.end() + db.redisClient.del("#{deviceId}_tasks", "#{deviceId}_faults", (err) -> + return throwError(err, response) if err + response.writeHead(200) + response.end() + ) ) ) ) @@ -335,15 +341,12 @@ 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) -> + db.redisClient.del("#{deviceId}_faults", (err) -> return throwError(err, response) if err - db.redisClient.del("#{deviceId}_no_tasks", "#{deviceId}_faults", (err) -> - return throwError(err, response) if err - response.writeHead(200) - response.end() - ) + response.writeHead(200) + response.end() ) ) )