diff --git a/acs.coffee b/acs.coffee index 4381fc4..4b66085 100644 --- a/acs.coffee +++ b/acs.coffee @@ -24,6 +24,7 @@ applyConfigurations = (currentRequest, taskList) -> if taskList.length util.log("#{currentRequest.deviceId}: Presets discrepancy found") apiFunctions.insertTasks(taskList, (err, tasks) -> + throw new Error(err) if err? task = tasks[0] util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})") runTask(currentRequest, task, {}) @@ -43,7 +44,7 @@ writeResponse = (currentRequest, res) -> if config.DEBUG_DEVICES[currentRequest.deviceId] dump = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n" fs = require('fs').appendFile("debug/#{currentRequest.deviceId}.dump", dump, (err) -> - throw new Error(err) if err + throw new Error(err) if err? ) currentRequest.httpResponse.writeHead(res.code, res.headers) @@ -102,14 +103,14 @@ updateDevice = (currentRequest, actions, callback) -> updates["#{path}_timestamp"] = now if Object.keys(updates).length > 0 or Object.keys(deletes).length > 0 - db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates, '$unset' : deletes}, {safe: true}, (err, count) -> + db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates, '$unset' : deletes}, {}, (err, count) -> if (err) callback(err) if callback? return if count == 0 util.log("#{currentRequest.deviceId}: New device detected") - db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {upsert: true, safe: true}, (err) -> + db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {upsert: true}, (err) -> if err? callback(err) if callback? return @@ -135,9 +136,10 @@ updateDevice = (currentRequest, actions, callback) -> runTask = (currentRequest, task, methodResponse) -> timeDiff = process.hrtime() tasks.task(task, methodResponse, (err, status, cwmpResponse, deviceUpdates) -> + throw new Error(err) if err? # TODO handle error updateDevice(currentRequest, deviceUpdates, (err) -> - util.error(err) if err? + throw new Error(err) if err? timeDiff = process.hrtime(timeDiff)[0] + 1 if timeDiff > 3 # in seconds @@ -148,23 +150,27 @@ runTask = (currentRequest, task, methodResponse) -> when tasks.STATUS_FINISHED db.memcached.del(String(task._id)) db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) -> + throw new Error(err) if err? util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})") nextTask(currentRequest) ) when tasks.STATUS_FAULT util.log("#{currentRequest.deviceId}: Fault response for task #{task._id}") db.saveTask(task, (err) -> + throw new Error(err) if err? # Faulty task. No more work to do until task is deleted. res = tr069.response(null, cwmpResponse) writeResponse(currentRequest, res) ) when tasks.STATUS_PENDING db.saveTask(task, (err) -> + throw new Error(err) if err? # task expects CPE confirmation later nextTask(currentRequest) ) when tasks.STATUS_STARTED db.updateTask(task, (err) -> + throw new Error(err) if err? res = tr069.response(task._id, cwmpResponse) writeResponse(currentRequest, res) ) @@ -184,11 +190,13 @@ isTaskExpired = (task) -> nextTask = (currentRequest) -> cur = db.tasksCollection.find({'device' : currentRequest.deviceId}).sort(['timestamp']).limit(1) cur.nextObject((err, task) -> + throw new Error(err) if err? cwmpResponse = {} if not task # no more tasks, check presets discrepancy db.memcached.get(["#{currentRequest.deviceId}_presets_hash", 'presets_hash'], (err, results) -> + # ignore memcached errors presetsHash = results['presets_hash'] devicePresetsHash = results["#{currentRequest.deviceId}_presets_hash"] @@ -218,6 +226,7 @@ nextTask = (currentRequest) -> writeResponse(currentRequest, res) else if isTaskExpired(task) db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) -> + throw new Error(err) if err? nextTask(currentRequest) ) else @@ -287,6 +296,7 @@ listener = (httpRequest, httpResponse) -> util.log("#{currentRequest.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") updateDevice(currentRequest, {'inform' : cwmpRequest.methodRequest.event, 'parameterValues' : cwmpRequest.methodRequest.parameterList}, (err) -> + throw new Error(err) if err? res = tr069.response(cwmpRequest.id, cwmpResponse, cookies) writeResponse(currentRequest, res) ) @@ -330,6 +340,8 @@ cluster = require 'cluster' numCPUs = require('os').cpus().length if cluster.isMaster + db.memcached.flush(() -> + ) cluster.on('listening', (worker, address) -> util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}") ) diff --git a/db.coffee b/db.coffee index dca4476..e9a4fd4 100644 --- a/db.coffee +++ b/db.coffee @@ -14,7 +14,8 @@ objectsCollection = null options = { db : { - native_parser: true + w : 1 + wtimeout : 60000 } server : { auto_reconnect : true @@ -24,24 +25,29 @@ options = { mongodb.MongoClient.connect("mongodb://#{config.MONGODB_SOCKET}/#{config.DATABASE_NAME}", options, (err, db) -> exports.mongoDb = db db.collection('tasks', (err, collection) -> + throw new Error(err) if err? exports.tasksCollection = tasksCollection = collection collection.ensureIndex({device: 1, timestamp: 1}, (err) -> ) ) db.collection('devices', (err, collection) -> + throw new Error(err) if err? exports.devicesCollection = devicesCollection = collection ) db.collection('presets', (err, collection) -> + throw new Error(err) if err? exports.presetsCollection = presetsCollection = collection ) db.collection('objects', (err, collection) -> + throw new Error(err) if err? exports.objectsCollection = objectsCollection = collection ) db.collection('fs.files', (err, collection) -> + throw new Error(err) if err? exports.filesCollection = filesCollection = collection ) ) @@ -56,6 +62,8 @@ getTask = (taskId, callback) -> task = data[tid] if not task? tasksCollection.findOne({_id : mongodb.ObjectID(tid)}, (err, task) -> + # TODO use err parameter in callback + throw new Error(err) if err? callback(task) ) else diff --git a/presets.coffee b/presets.coffee index 714f415..10b12e1 100644 --- a/presets.coffee +++ b/presets.coffee @@ -86,6 +86,7 @@ exports.assertPresets = (deviceId, presetsHash, callback) -> mongoQuery.optimizeProjection(projection) db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> + throw new Error(err) if err? devicePresets = [] for p in presets if mongoQuery.test(device, p.precondition) @@ -133,7 +134,9 @@ exports.assertPresets = (deviceId, presetsHash, callback) -> else if matchObject(objects[c.object], p) u = {} u["#{c.name}.#{k}._name"] = c.object - db.devicesCollection.update({'_id' : deviceId}, {'$set' : u}, {safe : false}) + db.devicesCollection.update({'_id' : deviceId}, {'$set' : u}, {}, (err, count) -> + throw new Error(err) if err? + ) instances[k] = p if Object.keys(instances).length > 0 @@ -163,10 +166,14 @@ exports.assertPresets = (deviceId, presetsHash, callback) -> log("#{deviceId}: Updating tags") if delete_tags.length > 0 - db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : {'$in' : delete_tags}}}, {safe : false}) + db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : {'$in' : delete_tags}}}, {}, (err, count) -> + throw new Error(err) if err? + ) if add_tags.length > 0 - db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : {'$each' : add_tags}}}, {safe : false}) + db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : {'$each' : add_tags}}}, {}, (err, count) -> + throw new Error(err) if err? + ) if getParameterValues.length taskList.push {device : deviceId, name : 'getParameterValues', parameterNames: getParameterValues, timestamp : new Date()}