From ee40a443396b6fae39f5d7c5bc16f17ed81590dd Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Fri, 23 Dec 2016 17:32:08 -0800 Subject: [PATCH] Remove use of domains for error handling --- lib/cache.coffee | 5 +- lib/cwmp.coffee | 56 ++++++++------- lib/nbi.coffee | 174 +++++++++++++++++++--------------------------- lib/server.coffee | 28 +++----- 4 files changed, 117 insertions(+), 146 deletions(-) diff --git a/lib/cache.coffee b/lib/cache.coffee index 2f4a78c..58e1755 100644 --- a/lib/cache.coffee +++ b/lib/cache.coffee @@ -92,6 +92,7 @@ refresh = (callback) -> return callback() lock('presets_hash_lock', 3000, (err, unlockOrExtend) -> + return callback(err) if err counter = 4 db.presetsCollection.find().toArray((err, res) -> if err @@ -145,7 +146,9 @@ refresh = (callback) -> when 'provision' _provisions.push([c.name].concat(c.args or [])) else - throw new Error("Unknown configuration type #{c.type}") + callback(new Error("Unknown configuration type #{c.type}")) if -- counter >= 0 + counter = 0 + return presets.push({name: preset._id, channel: preset.channel or 'default', schedule: schedule, events: events, precondition: precondition, provisions: _provisions}) diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index 02f4f00..1f6a01b 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -53,11 +53,19 @@ cache = require './cache' db = require './db' +throwError = (err, httpResponse) -> + if httpResponse + httpResponse.writeHead(500, {'Connection' : 'close'}) + httpResponse.end("#{err.name}: #{err.message}") + + throw err + + writeResponse = (currentRequest, res) -> if config.get('DEBUG', currentRequest.sessionData.deviceId) dump = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n" require('fs').appendFile("./debug/#{currentRequest.sessionData.deviceId}.dump", dump, (err) -> - throw err if err + throwError(err) if err ) # respond using the same content-encoding as the request @@ -72,7 +80,7 @@ writeResponse = (currentRequest, res) -> if compress? compress(res.data, (err, data) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err res.headers['Content-Length'] = data.length currentRequest.httpResponse.writeHead(res.code, res.headers) currentRequest.httpResponse.end(data) @@ -123,7 +131,7 @@ inform = (currentRequest, cwmpRequest) -> util.log("#{currentRequest.sessionData.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") session.inform(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err res = soap.response({ id : cwmpRequest.id, methodResponse : rpcResponse, @@ -141,7 +149,7 @@ inform = (currentRequest, cwmpRequest) -> transferComplete = (currentRequest, cwmpRequest) -> session.transferComplete(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse, fault, operation) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err if fault for k, v of operation.retries @@ -207,7 +215,7 @@ appendProvisions = (original, toAppend) -> applyPresets = (currentRequest) -> cache.getPresets((err, presetsHash, presets) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err # Filter presets based on existing faults blackList = {} @@ -267,7 +275,7 @@ applyPresets = (currentRequest) -> declarations.push([v, 1, {value: 1}]) session.rpcRequest(currentRequest.sessionData, declarations, (err, fault, id, rpcRequest) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err if fault recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels) @@ -304,7 +312,7 @@ applyPresets = (currentRequest) -> currentRequest.sessionData.faultsTouched[channel] = true session.rpcRequest(currentRequest.sessionData, null, (err, fault, id, rpcRequest) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err if fault recordFault(currentRequest.sessionData, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels) @@ -329,7 +337,7 @@ applyPresets = (currentRequest) -> nextRpc = (currentRequest) -> session.rpcRequest(currentRequest.sessionData, null, (err, fault, id, rpcRequest) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err if fault for channel of currentRequest.sessionData.channels @@ -392,7 +400,7 @@ nextRpc = (currentRequest) -> 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') + return throwError(new Error('Task name not recognized'), currentRequest.httpResponse) if err return nextRpc(currentRequest) @@ -403,14 +411,14 @@ nextRpc = (currentRequest) -> sendRpcRequest = (currentRequest, id, rpcRequest) -> if not rpcRequest? session.end(currentRequest.sessionData, (err, isNew) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err delete currentSessions.get(currentRequest.httpRequest.connection)[currentRequest.sessionData.sessionId] if isNew util.log("#{currentRequest.sessionData.deviceId}: New device registered") db.clearTasks(currentRequest.sessionData.doneTasks, (err) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err counter = 1 @@ -421,7 +429,7 @@ sendRpcRequest = (currentRequest, id, rpcRequest) -> db.saveFault(currentRequest.sessionData.deviceId, k, currentRequest.sessionData.faults[k], (err) -> -- counter if err and counter >= 0 - throw err + return throwError(err, currentRequest.httpResponse) if err if counter == 0 return writeResponse(currentRequest, soap.response(null)) @@ -430,7 +438,7 @@ sendRpcRequest = (currentRequest, id, rpcRequest) -> db.deleteFault(currentRequest.sessionData.deviceId, k, (err) -> -- counter if err and counter >= 0 - throw err + return throwError(err, currentRequest.httpResponse) if err if counter == 0 return writeResponse(currentRequest, soap.response(null)) @@ -454,7 +462,7 @@ sendRpcRequest = (currentRequest, id, rpcRequest) -> if not rpcRequest.fileSize? return cache.getFiles((err, hash, files) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err if rpcRequest.fileName of files rpcRequest.fileSize = files[rpcRequest.fileName].length @@ -510,10 +518,10 @@ onConnection = (socket) -> sessions = currentSessions.get(socket) for sessionId, sessionData of sessions session.serialize(sessionData, (err, sessionDataString) -> - throw err if err + return throwError(err) if err # TODO don't set if process is shutting down db.redisClient.setex("session_#{sessionId}", sessionData.timeout, sessionDataString, (err) -> - throw err if err + return throwError(err) if err ) ) ) @@ -559,7 +567,7 @@ listener = (httpRequest, httpResponse) -> stream.on('end', () -> cwmpRequest = null getSession(httpRequest, f = (err, sessionData) -> - throw err if err + return throwError(err, httpResponse) if err cwmpRequest ?= soap.request(httpRequest, sessionData?.cwmpVersion) if not sessionData? if cwmpRequest.methodRequest?.type isnt 'Inform' @@ -569,7 +577,7 @@ listener = (httpRequest, httpResponse) -> deviceId = common.generateDeviceId(cwmpRequest.methodRequest.deviceId) return session.init(deviceId, cwmpRequest.cwmpVersion, cwmpRequest.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId), (err, sessionData) -> - throw err if err + return throwError(err, httpResponse) if err sessionData.sessionId = crypto.randomBytes(8).toString('hex') @@ -578,7 +586,7 @@ listener = (httpRequest, httpResponse) -> httpRequest.connection.setTimeout(sessionData.timeout * 1000) db.getDueTasksAndFaultsAndOperations(deviceId, sessionData.timestamp, (err, dueTasks, faults, operations) -> - throw err if err + return throwError(err, httpResponse) if err sessionData.tasks = dueTasks sessionData.faults = faults sessionData.retries = {} @@ -606,7 +614,7 @@ listener = (httpRequest, httpResponse) -> if config.get('DEBUG', currentRequest.sessionData.deviceId) dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{httpRequest.getBody()}\n\n" require('fs').appendFile("./debug/#{currentRequest.sessionData.deviceId}.dump", dump, (err) -> - throw err if err + return throwError(err) if err ) if cwmpRequest.methodRequest? @@ -623,15 +631,15 @@ listener = (httpRequest, httpResponse) -> }) writeResponse(currentRequest, res) else if cwmpRequest.methodRequest.type is 'TransferComplete' - throw new Error('ACS method not supported') + return throwError(new Error('ACS method not supported'), currentRequest.httpResponse) if err else if cwmpRequest.methodResponse? session.rpcResponse(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.methodResponse, (err) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err nextRpc(currentRequest) ) else if cwmpRequest.fault? session.rpcFault(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.fault, (err, fault) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err if fault for channel of currentRequest.sessionData.channels @@ -649,7 +657,7 @@ listener = (httpRequest, httpResponse) -> ) else # CPE sent empty response session.timeoutOperations(currentRequest.sessionData, (err, faults, operations) -> - throw err if err + return throwError(err, currentRequest.httpResponse) if err for fault, i in faults for k, v of operations[i].retries diff --git a/lib/nbi.coffee b/lib/nbi.coffee index 5c56e17..8daac24 100644 --- a/lib/nbi.coffee +++ b/lib/nbi.coffee @@ -63,8 +63,13 @@ VIRTUAL_PARAMETERS_REGEX = /^\/virtual_parameters\/([a-zA-Z0-9\-\_\%]+)\/?$/ FAULTS_REGEX = /^\/faults\/([a-zA-Z0-9\-\_\%\:]+)\/?$/ -errorToString = (err) -> - "#{err.name}: #{err.message}" + +throwError = (err, httpResponse) -> + if httpResponse + httpResponse.writeHead(500, {'Connection' : 'close'}) + httpResponse.end("#{err.name}: #{err.message}") + + throw err listener = (request, response) -> @@ -95,27 +100,21 @@ listener = (request, response) -> preset._id = presetName db.presetsCollection.save(preset, (err) -> + return throwError(err, response) if err db.redisClient.del('presets_hash', (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else if request.method == 'DELETE' db.presetsCollection.remove({'_id' : presetName}, (err) -> + return throwError(err, response) if err db.redisClient.del('presets_hash', (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else response.writeHead 405, {'Allow': 'PUT, DELETE'} @@ -127,27 +126,21 @@ listener = (request, response) -> object._id = objectName db.objectsCollection.save(object, (err) -> + return throwError(err, response) if err db.redisClient.del('objects', 'presets_hash', (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else if request.method == 'DELETE' db.objectsCollection.remove({'_id' : objectName}, (err) -> + return throwError(err, response) if err db.redisClient.del('objects', 'presets_hash', (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else response.writeHead 405, {'Allow': 'PUT, DELETE'} @@ -161,27 +154,21 @@ listener = (request, response) -> } db.provisionsCollection.save(object, (err) -> + return throwError(err, response) if err db.redisClient.del('presets_hash', (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else if request.method == 'DELETE' db.provisionsCollection.remove({'_id' : provisionName}, (err) -> + return throwError(err, response) if err db.redisClient.del('presets_hash', (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else response.writeHead 405, {'Allow': 'PUT, DELETE'} @@ -195,27 +182,21 @@ listener = (request, response) -> } db.virtualParametersCollection.save(object, (err) -> + return throwError(err, response) if err db.redisClient.del('presets_hash', (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else if request.method == 'DELETE' db.virtualParametersCollection.remove({'_id' : virtualParameterName}, (err) -> + return throwError(err, response) if err db.redisClient.del('presets_hash', (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else response.writeHead 405, {'Allow': 'PUT, DELETE'} @@ -226,27 +207,21 @@ listener = (request, response) -> tag = querystring.unescape(r[2]) if request.method == 'POST' db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : tag}}, {safe: true}, (err) -> + return throwError(err, response) if err db.redisClient.del("#{deviceId}_presets_hash", (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else if request.method == 'DELETE' db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : tag}}, {safe: true}, (err) -> + return throwError(err, response) if err db.redisClient.del("#{deviceId}_presets_hash", (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) - if err - response.writeHead(500) - response.end(errorToString(err)) - return - response.writeHead(200) - response.end() ) else response.writeHead 405, {'Allow': 'POST, DELETE'} @@ -257,12 +232,16 @@ listener = (request, response) -> 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) -> - throw err if err + return throwError(err, response) if err if channel.startsWith('_task_') return db.tasksCollection.remove({_id : new mongodb.ObjectID(channel.slice(6))}, (err) -> - throw err if err + return throwError(err, response) if err + response.writeHead(200) + response.end() ) + response.writeHead(200) response.end() ) @@ -277,9 +256,9 @@ listener = (request, response) -> task = JSON.parse(body) task.device = deviceId apiFunctions.insertTasks(task, (err) -> - throw err if err + return throwError(err, response) if err db.redisClient.del("#{deviceId}_no_tasks", (err) -> - throw err if err + return throwError(err, response) if err if urlParts.query.connection_request? apiFunctions.connectionRequest(deviceId, (err) -> @@ -288,20 +267,14 @@ listener = (request, response) -> response.end(JSON.stringify(task)) else apiFunctions.watchTask(task._id, config.get('DEVICE_ONLINE_THRESHOLD', deviceId), (err, status) -> - if err - response.writeHead(500) - response.end(errorToString(err)) - return + return throwError(err, response) if err if status is 'timeout' response.writeHead(202, 'Task queued but not processed', {'Content-Type' : 'application/json'}) response.end(JSON.stringify(task)) else if status is 'fault' db.tasksCollection.findOne({_id : task._id}, (err, task) -> - if err - response.writeHead(500) - response.end(errorToString(err)) - return + return throwError(err, response) if err response.writeHead(202, 'Task faulted', {'Content-Type' : 'application/json'}) response.end(JSON.stringify(task)) @@ -321,7 +294,7 @@ listener = (request, response) -> apiFunctions.connectionRequest(deviceId, (err) -> if err response.writeHead 504 - response.end(errorToString(err)) + response.end("#{err.name}: #{err.message}") return response.writeHead 200 response.end() @@ -339,12 +312,12 @@ listener = (request, response) -> if not action? or action is '/' if request.method == 'DELETE' db.tasksCollection.findOne({'_id' : taskId}, {'device' : 1}, (err, task) -> - throw err if err + return throwError(err, response) if err deviceId = task.device db.tasksCollection.remove({'_id' : taskId}, (err) -> - throw err if err + return throwError(err, response) if err db.faultsCollection.remove({_id : "#{deviceId}:_task_#{String(taskId)}"}, (err) -> - throw err if err + return throwError(err, response) if err response.writeHead(200) response.end() ) @@ -356,14 +329,14 @@ listener = (request, response) -> else if action is '/retry' if request.method == 'POST' db.tasksCollection.findOne({'_id' : taskId}, {'device' : 1}, (err, task) -> - throw err if err + return throwError(err, response) if err deviceId = task.device db.tasksCollection.update({_id : taskId}, {$unset : {fault : 1}, $set : {timestamp : new Date()}}, (err, count) -> - throw err if err + return throwError(err, response) if err db.faultsCollection.remove({_id : "#{deviceId}:_task_#{String(taskId)}"}, (err) -> - throw err if err + return throwError(err, response) if err db.redisClient.del("#{deviceId}_no_tasks", "#{deviceId}_faults", (err) -> - throw err if err + return throwError(err, response) if err response.writeHead(200) response.end() ) @@ -389,9 +362,9 @@ listener = (request, response) -> gs = new mongodb.GridStore(db.mongoDb, filename, filename, 'w', {metadata : metadata}) gs.open((err, gs) -> gs.write(body, (err, res) -> - throw err if err + return throwError(err, response) if err gs.close((err) -> - throw err if err + return throwError(err, response) if err response.writeHead(201) response.end() ) @@ -410,7 +383,7 @@ listener = (request, response) -> require('child_process').exec("ping -w 1 -i 0.2 -c 3 #{host}", (err, stdout, stderr) -> if err response.writeHead(404, {'Cache-Control' : 'no-cache'}) - response.end(errorToString(err)) + response.end("#{err.name}: #{err.message}") return response.writeHead(200, {'Content-Type' : 'text/plain', 'Cache-Control' : 'no-cache'}) response.end(stdout) @@ -423,10 +396,7 @@ listener = (request, response) -> deviceId = querystring.unescape(DELETE_DEVICE_REGEX.exec(urlParts.pathname)[1]) apiFunctions.deleteDevice(deviceId, (err) -> - if err - response.writeHead(500) - response.end(errorToString(err)) - return + return throwError(err, response) if err response.writeHead(200) response.end() ) @@ -456,7 +426,7 @@ listener = (request, response) -> q = JSON.parse(urlParts.query.query) catch err response.writeHead(400) - response.end(errorToString(err)) + response.end("#{err.name}: #{err.message}") return else q = {} @@ -503,7 +473,9 @@ listener = (request, response) -> response.write("[\n") i = 0 cur.each((err, item) -> - throw err if err + if err + throwError(err, response) + return false if item? response.write(",\n") if i++ diff --git a/lib/server.coffee b/lib/server.coffee index 183356e..b7aa33a 100644 --- a/lib/server.coffee +++ b/lib/server.coffee @@ -15,7 +15,6 @@ # along with GenieACS. If not, see . ### -domain = require 'domain' util = require 'util' cluster = require 'cluster' config = require './config' @@ -31,7 +30,7 @@ networkInterface = config.get("#{service.toUpperCase()}_INTERFACE") port = config.get("#{service.toUpperCase()}_PORT") useHttps = config.get("#{service.toUpperCase()}_SSL") -serviceListener = require("./#{service}").listener +listener = require("./#{service}").listener onConnection = require("./#{service}").onConnection server = null @@ -48,25 +47,14 @@ exit = () -> ) -listener = (httpRequest, httpResponse) -> - d = domain.create() - - d.on('error', (err) -> +process.on('uncaughtException', (err) -> + try util.error("#{new Date().toISOString()} - #{err.stack}\n") - try - exit() - httpResponse.writeHead(500, {'Connection' : 'close'}) - httpResponse.end("#{err.name}: #{err.message}") - catch err2 - util.error("#{new Date().toISOString()} - #{err2.stack}\n") - ) - - d.add(httpRequest) - d.add(httpResponse) - - d.run(() -> - serviceListener(httpRequest, httpResponse) - ) + exit() + catch err + util.error("#{new Date().toISOString()} - #{err2.stack}\n") + throw err +) if useHttps