Remove use of domains for error handling

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})
 

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

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,28 +100,22 @@ 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
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             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
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             response.writeHead(200)
             response.end()
           )
+        )
       else
         response.writeHead 405, {'Allow': 'PUT, DELETE'}
         response.end('405 Method Not Allowed')
@@ -127,28 +126,22 @@ 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
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             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
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             response.writeHead(200)
             response.end()
           )
+        )
       else
         response.writeHead 405, {'Allow': 'PUT, DELETE'}
         response.end('405 Method Not Allowed')
@@ -161,28 +154,22 @@ listener = (request, response) ->
         }
 
         db.provisionsCollection.save(object, (err) ->
+          return throwError(err, response) if err
           db.redisClient.del('presets_hash', (err) ->
-            throw err if err
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             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
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             response.writeHead(200)
             response.end()
           )
+        )
       else
         response.writeHead 405, {'Allow': 'PUT, DELETE'}
         response.end('405 Method Not Allowed')
@@ -195,28 +182,22 @@ listener = (request, response) ->
         }
 
         db.virtualParametersCollection.save(object, (err) ->
+          return throwError(err, response) if err
           db.redisClient.del('presets_hash', (err) ->
-            throw err if err
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             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
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             response.writeHead(200)
             response.end()
           )
+        )
       else
         response.writeHead 405, {'Allow': 'PUT, DELETE'}
         response.end('405 Method Not Allowed')
@@ -226,28 +207,22 @@ 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
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             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
-          )
-          if err
-            response.writeHead(500)
-            response.end(errorToString(err))
-            return
+            return throwError(err, response) if err
             response.writeHead(200)
             response.end()
           )
+        )
       else
         response.writeHead 405, {'Allow': 'POST, DELETE'}
         response.end('405 Method Not Allowed')
@@ -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++

lib/server.coffee

@@ -15,7 +15,6 @@
 # along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
 ###
 
-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,24 +47,13 @@ exit = () ->
   )
 
 
-listener = (httpRequest, httpResponse) ->
-  d = domain.create()
-
-  d.on('error', (err) ->
-    util.error("#{new Date().toISOString()} - #{err.stack}\n")
+process.on('uncaughtException', (err) ->
   try
+    util.error("#{new Date().toISOString()} - #{err.stack}\n")
     exit()
-      httpResponse.writeHead(500, {'Connection' : 'close'})
-      httpResponse.end("#{err.name}: #{err.message}")
-    catch err2
+  catch err
     util.error("#{new Date().toISOString()} - #{err2.stack}\n")
-  )
-
-  d.add(httpRequest)
-  d.add(httpResponse)
-
-  d.run(() ->
-    serviceListener(httpRequest, httpResponse)
+  throw err
 )