Convert cwmp.coffee to JS

.gitignore

@@ -1,6 +1,5 @@
 *~
 lib/*.map
-lib/cwmp.js
 lib/session.js
 node_modules
 debug

lib/cwmp.coffee

@@ -1,965 +0,0 @@
-###
-# Copyright 2013-2017  Zaid Abdulla
-#
-# This file is part of GenieACS.
-#
-# GenieACS is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# GenieACS is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Affero General Public License for more details.
-#
-# You should have received a copy of the GNU Affero General Public License
-# along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
-###
-
-util = require 'util'
-zlib = require 'zlib'
-crypto = require 'crypto'
-
-config = require './config'
-common = require './common'
-soap = require './soap'
-session = require './session'
-query = require './query'
-device = require './device'
-cache = require './cache'
-localCache = require './local-cache'
-db = require './db'
-logger = require './logger'
-scheduling = require './scheduling'
-
-
-MAX_CYCLES = 4
-
-MAX_CONCURRENT_REQUESTS = config.get('MAX_CONCURRENT_REQUESTS')
-stats = {
-  concurrentRequests: 0
-  totalRequests: 0
-  droppedRequests: 0
-  initiatedSessions: 0
-}
-
-throwError = (err, httpResponse) ->
-  if httpResponse
-    currentSessions.delete(httpResponse.connection)
-    httpResponse.writeHead(500, {'Connection' : 'close'})
-    httpResponse.end("#{err.name}: #{err.message}")
-    stats.concurrentRequests -= 1
-
-  throw err
-
-
-writeResponse = (sessionContext, res, close) ->
-  # Close connection after last request in session
-  res.headers['Connection'] = 'close' if close
-
-  if config.get('DEBUG', sessionContext.deviceId)
-    dump = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n"
-    require('fs').appendFile("./debug/#{sessionContext.deviceId}.dump", dump, (err) ->
-      throwError(err) if err
-    )
-
-  finish = () ->
-    stats.concurrentRequests -= 1
-    if sessionContext.httpRequest.connection.destroyed
-      logger.accessError({
-        sessionContext: sessionContext
-        message: 'Connection dropped'
-      })
-    else if close
-      endSession(sessionContext, (err, isNew) ->
-        return throwError(err) if err
-
-        if isNew
-          logger.accessInfo({
-            sessionContext: sessionContext
-            message: 'New device registered'
-          })
-      )
-    else
-      sessionContext.lastActivity = Date.now()
-      currentSessions.set(sessionContext.httpRequest.connection, sessionContext)
-
-  # respond using the same content-encoding as the request
-  if sessionContext.httpRequest.headers['content-encoding']? and res.data.length > 0
-    switch sessionContext.httpRequest.headers['content-encoding']
-      when 'gzip'
-        res.headers['Content-Encoding'] = 'gzip'
-        compress = zlib.gzip
-      when 'deflate'
-        res.headers['Content-Encoding'] = 'deflate'
-        compress = zlib.deflate
-
-  if compress?
-    compress(res.data, (err, data) ->
-      return throwError(err, sessionContext.httpResponse) if err
-      res.headers['Content-Length'] = data.length
-      sessionContext.httpResponse.writeHead(res.code, res.headers)
-      sessionContext.httpResponse.end(data)
-      finish()
-    )
-  else
-    res.headers['Content-Length'] = res.data.length
-    sessionContext.httpResponse.writeHead(res.code, res.headers)
-    sessionContext.httpResponse.end(res.data)
-    finish()
-
-
-recordFault = (sessionContext, fault, provisions, channels) ->
-  if not provisions
-    provisions = sessionContext.provisions
-    channels = sessionContext.channels
-
-  faults = sessionContext.faults
-
-  for channel of channels
-    provs = sessionContext.faults[channel]?.provisions or []
-    faults[channel] = Object.assign({provisions: provs}, fault)
-    if channel.startsWith('task_')
-      taskId = channel.slice(5)
-      for t in sessionContext.tasks
-        faults[channel].expiry = t.expiry if t._id == taskId and t.expiry
-
-    if sessionContext.retries[channel]?
-      ++ sessionContext.retries[channel]
-    else
-      sessionContext.retries[channel] = 0
-      if Object.keys(channels).length != 1
-        faults[channel].retryNow = true
-
-    if channels[channel] == 0
-      faults[channel].precondition = true
-
-    sessionContext.faultsTouched ?= {}
-    sessionContext.faultsTouched[channel] = true
-
-    logger.accessWarn({
-      sessionContext: sessionContext
-      message: 'Channel has faulted'
-      fault: fault
-      channel: channel
-      retries: sessionContext.retries[channel]
-    })
-
-  for provision, i in provisions
-    for channel of channels
-      if (channels[channel] >> i) & 1
-        faults[channel].provisions.push(provision)
-
-  for channel of channels
-    provs = faults[channel].provisions
-    faults[channel].provisions = []
-    appendProvisions(faults[channel].provisions, provs)
-
-  session.clearProvisions(sessionContext)
-  return
-
-
-inform = (sessionContext, rpc) ->
-  session.inform(sessionContext, rpc.cpeRequest, (err, acsResponse) ->
-    return throwError(err, sessionContext.httpResponse) if err
-    res = soap.response({
-      id: rpc.id,
-      acsResponse: acsResponse,
-      cwmpVersion : sessionContext.cwmpVersion
-    })
-
-    if !!cookiesPath = config.get('COOKIES_PATH', sessionContext.deviceId)
-      res.headers['Set-Cookie'] = "session=#{sessionContext.sessionId}; Path=#{cookiesPath}"
-    else
-      res.headers['Set-Cookie'] = "session=#{sessionContext.sessionId}"
-
-    writeResponse(sessionContext, res)
-  )
-
-
-transferComplete = (sessionContext, rpc) ->
-  session.transferComplete(sessionContext, rpc.cpeRequest, (err, acsResponse, operation, fault) ->
-    return throwError(err, sessionContext.httpResponse) if err
-
-    if not operation?
-      logger.accessWarn({
-        sessionContext: sessionContext
-        message: 'Unrecognized command key'
-        rpc: rpc
-      })
-
-    if fault
-      for k, v of operation.retries
-        sessionContext.retries[k] = v
-
-      recordFault(sessionContext, fault, operation.provisions, operation.channels)
-
-    res = soap.response({
-      id : rpc.id,
-      acsResponse : acsResponse,
-      cwmpVersion : sessionContext.cwmpVersion
-    })
-    writeResponse(sessionContext, res)
-  )
-
-
-# Append providions and remove duplicates
-appendProvisions = (original, toAppend) ->
-  modified = false
-  stringified = new WeakMap()
-  for p, i in original
-    stringified.set(p, JSON.stringify(p))
-
-  for p, i in toAppend by -1
-    s = JSON.stringify(p)
-    for pp, j in original by -1
-      ss = stringified.get(pp)
-      if s == ss
-        if (not p) or j >= original.length - (toAppend.length - i)
-          p = null
-        else
-          original.splice(j, 1)
-          modified = true
-    if p
-      original.splice(original.length - (toAppend.length - i) + 1, 0, p)
-      stringified.set(p, s)
-      modified = true
-
-  return modified
-
-
-applyPresets = (sessionContext) ->
-  localCache.getPresets((err, presetsHash, presets) ->
-    return throwError(err, sessionContext.httpResponse) if err
-
-    # Filter presets based on existing faults
-    blackList = {}
-    whiteList = null
-    whiteListProvisions = null
-
-    RETRY_DELAY = config.get('RETRY_DELAY', sessionContext.deviceId)
-
-    if sessionContext.faults?
-      for channel, fault of sessionContext.faults
-        if fault.retryNow
-          retryTimestamp = 0
-        else
-          retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, sessionContext.retries[channel])) * 1000
-
-        if retryTimestamp <= sessionContext.timestamp
-          whiteList = channel
-          whiteListProvisions = fault.provisions
-          break
-
-        blackList[channel] = if fault.precondition then 1 else 2
-
-    sessionContext.deviceData.timestamps.revision = 1
-    sessionContext.deviceData.attributes.revision = 1
-
-    deviceEvents = {}
-    for p in sessionContext.deviceData.paths.find(['Events', '*'], false, true)
-      if sessionContext.timestamp <= sessionContext.deviceData.attributes.get(p)?.value?[1][0]
-        deviceEvents[p[1]] = true
-
-    parameters = {}
-    filteredPresets = []
-
-    for preset in presets
-      if whiteList?
-        continue if preset.channel != whiteList
-      else if blackList[preset.channel] == 1
-        continue
-
-      eventsMatch = true
-      for k, v of preset.events
-        if (!v) != (!deviceEvents[k.replace(' ', '_')])
-          eventsMatch = false
-          break
-
-      continue if not eventsMatch
-
-      if preset.schedule
-        continue if not preset.schedule.schedule
-        r = scheduling.cron(sessionContext.timestamp, preset.schedule.schedule)
-        if not (r[0] + preset.schedule.duration > sessionContext.timestamp)
-          continue
-
-      filteredPresets.push(preset)
-      for k of preset.precondition
-        sessionContext.channels[preset.channel] = 0
-        p = k.split(/([^a-zA-Z0-9\-\_\.].*)/, 1)[0]
-        parameters[p] = common.parsePath(p)
-
-    declarations = []
-    for k, v of parameters
-      declarations.push([v, 1, {value: 1}])
-
-    session.rpcRequest(sessionContext, declarations, (err, fault, id, acsRequest) ->
-      return throwError(err, sessionContext.httpResponse) if err
-
-      if fault
-        recordFault(sessionContext, fault)
-        session.clearProvisions(sessionContext)
-        return applyPresets(sessionContext)
-
-      if acsRequest
-        return sendAcsRequest(sessionContext, id, acsRequest)
-
-      session.clearProvisions(sessionContext)
-
-      for k, v of parameters
-        unpacked = device.unpack(sessionContext.deviceData, v)
-        if unpacked[0] and (vv = sessionContext.deviceData.attributes.get(unpacked[0]).value?[1])?
-          parameters[k] = vv[0]
-        else
-          delete parameters[k]
-
-      if whiteList?
-        session.addProvisions(sessionContext, whiteList, whiteListProvisions)
-
-      appendProvisionsToFaults = {}
-      for p in filteredPresets
-        if query.testFilter(parameters, p.precondition)
-          if blackList[p.channel] == 2
-            appendProvisionsToFaults[p.channel] =
-              (appendProvisionsToFaults[p.channel] or []).concat(p.provisions)
-          else
-            session.addProvisions(sessionContext, p.channel, p.provisions)
-
-      for channel, provisions of appendProvisionsToFaults
-        if appendProvisions(sessionContext.faults[channel].provisions, provisions)
-          sessionContext.faultsTouched ?= {}
-          sessionContext.faultsTouched[channel] = true
-
-      # Don't increment when processing individual channels (e.g. after fault)
-      sessionContext.presetCycles = (sessionContext.presetCycles or 0) + 1 if not whiteList?
-
-      if sessionContext.presetCycles > MAX_CYCLES
-        fault = {
-          code: 'preset_loop'
-          message: 'The presets are stuck in an endless configuration loop'
-          timestamp: sessionContext.timestamp
-        }
-        recordFault(sessionContext, fault)
-        # No need to save retryNow
-        for c of sessionContext.faults
-          delete sessionContext.faults[c].retryNow
-        session.clearProvisions(sessionContext)
-        return sendAcsRequest(sessionContext)
-
-      sessionContext.deviceData.timestamps.dirty = 0;
-      sessionContext.deviceData.attributes.dirty = 0;
-
-      session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) ->
-        return throwError(err, sessionContext.httpResponse) if err
-
-        if fault
-          recordFault(sessionContext, fault)
-          session.clearProvisions(sessionContext)
-          return applyPresets(sessionContext)
-
-        if not acsRequest
-          for channel, flags of sessionContext.channels
-            if channel of sessionContext.faults
-              delete sessionContext.faults[channel]
-              sessionContext.faultsTouched ?= {}
-              sessionContext.faultsTouched[channel] = true
-
-          if whiteList?
-            return applyPresets(sessionContext)
-
-          if sessionContext.deviceData.timestamps.dirty > 1 or
-              sessionContext.deviceData.attributes.dirty > 1
-            return applyPresets(sessionContext)
-
-        sendAcsRequest(sessionContext, id, acsRequest)
-      )
-    )
-  )
-
-
-nextRpc = (sessionContext) ->
-  session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) ->
-    return throwError(err, sessionContext.httpResponse) if err
-
-    if fault
-      recordFault(sessionContext, fault)
-      session.clearProvisions(sessionContext)
-      return nextRpc(sessionContext)
-
-    if acsRequest
-      return sendAcsRequest(sessionContext, id, acsRequest)
-
-    for channel, flags of sessionContext.channels
-      if flags and channel of sessionContext.faults
-        delete sessionContext.faults[channel]
-        sessionContext.faultsTouched ?= {}
-        sessionContext.faultsTouched[channel] = true
-
-      if channel.startsWith('task_')
-        taskId = channel.slice(5)
-        sessionContext.doneTasks ?= []
-        sessionContext.doneTasks.push(taskId)
-        for t, j in sessionContext.tasks
-          if t._id == taskId
-            sessionContext.tasks.splice(j, 1)
-            break
-
-    session.clearProvisions(sessionContext)
-
-    # Clear expired tasks
-    sessionContext.tasks = sessionContext.tasks.filter((task) ->
-      return true if not (task.expiry <= sessionContext.timestamp)
-
-      logger.accessInfo({
-        sessionContext: sessionContext
-        message: 'Task expired'
-        task: task
-      })
-      sessionContext.doneTasks ?= []
-      sessionContext.doneTasks.push(task._id)
-      channel = "task_#{task._id}"
-      if channel of sessionContext.faults
-        delete sessionContext.faults[channel]
-        sessionContext.faultsTouched ?= {}
-        sessionContext.faultsTouched[channel] = true
-
-      return false
-    )
-
-    task = sessionContext.tasks.find((task) -> !sessionContext.faults["task_#{task._id}"])
-
-    return applyPresets(sessionContext) if not task
-
-    switch task.name
-      when 'getParameterValues'
-        # Set channel in case params array is empty
-        sessionContext.channels["task_#{task._id}"] = 0
-        for p in task.parameterNames
-          session.addProvisions(sessionContext, "task_#{task._id}",
-            [['refresh', p]])
-      when 'setParameterValues'
-        # Set channel in case params array is empty
-        sessionContext.channels["task_#{task._id}"] = 0
-        for p in task.parameterValues
-          session.addProvisions(sessionContext, "task_#{task._id}",
-            [['value', p[0], p[1]]])
-      when 'refreshObject'
-        session.addProvisions(sessionContext, "task_#{task._id}",
-          [['refresh', task.objectName]])
-      when 'reboot'
-        session.addProvisions(sessionContext, "task_#{task._id}",
-          [['reboot']])
-      when 'factoryReset'
-        session.addProvisions(sessionContext, "task_#{task._id}",
-          [['reset']])
-      when 'download'
-        session.addProvisions(sessionContext, "task_#{task._id}",
-          [['download', task.fileType, task.fileName, task.targetFileName]])
-      when 'addObject'
-        alias = ("#{p[0]}:#{JSON.stringify(p[1])}" for p in task.parameterValues or []).join(',')
-        session.addProvisions(sessionContext, "task_#{task._id}",
-          [['instances', "#{task.objectName}.[#{alias}]", "+1"]]);
-      when 'deleteObject'
-        session.addProvisions(sessionContext, "task_#{task._id}",
-          [['instances', task.objectName, 0]]);
-      else
-        return throwError(new Error('Task name not recognized'), sessionContext.httpResponse)
-
-    return nextRpc(sessionContext)
-  )
-
-
-endSession = (sessionContext, callback) ->
-  saveCache = sessionContext.cacheUntil?
-
-  counter = 3
-
-  counter += 2
-  db.saveDevice(sessionContext.deviceId, sessionContext.deviceData, sessionContext.new, sessionContext.timestamp, (err) ->
-    if err
-      callback(err) if counter & 1
-      return counter = 0
-
-    return callback(null, sessionContext.new) if (counter -= 2) == 1
-  )
-
-  for k of sessionContext.operationsTouched
-    counter += 2
-    saveCache = true
-    if sessionContext.operations[k]?
-      db.saveOperation(sessionContext.deviceId, k, sessionContext.operations[k], (err) ->
-        if err
-          callback(err) if counter & 1
-          return counter = 0
-
-        return callback(null, sessionContext.new) if (counter -= 2) == 1
-      )
-    else
-      db.deleteOperation(sessionContext.deviceId, k, (err) ->
-        if err
-          callback(err) if counter & 1
-          return counter = 0
-
-        return callback(null, sessionContext.new) if (counter -= 2) == 1
-      )
-
-  if sessionContext.doneTasks?.length
-    counter += 2
-    saveCache = true
-    db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, (err) ->
-      if err
-        callback(err) if counter & 1
-        return counter = 0
-
-      return callback(null, sessionContext.new) if (counter -= 2) == 1
-    )
-
-  for k of sessionContext.faultsTouched
-    counter += 2
-    saveCache = true
-    if sessionContext.faults[k]
-      sessionContext.faults[k].retries = sessionContext.retries[k]
-      db.saveFault(sessionContext.deviceId, k, sessionContext.faults[k], (err) ->
-        if err
-          callback(err) if counter & 1
-          return counter = 0
-
-        return callback(null, sessionContext.new) if (counter -= 2) == 1
-      )
-    else
-      db.deleteFault(sessionContext.deviceId, k, (err) ->
-        if err
-          callback(err) if counter & 1
-          return counter = 0
-
-        return callback(null, sessionContext.new) if (counter -= 2) == 1
-      )
-
-  if saveCache
-    counter += 2
-    cacheDueTasksAndFaultsAndOperations(sessionContext.deviceId, sessionContext.tasks, sessionContext.faults, sessionContext.operations, sessionContext.cacheUntil, (err) ->
-      if err
-        callback(err) if counter & 1
-        return counter = 0
-
-      return callback(null, sessionContext.new) if (counter -= 2) == 1
-    )
-
-  callback(null, sessionContext.new) if (counter -= 2) == 1
-
-
-sendAcsRequest = (sessionContext, id, acsRequest) ->
-  if not acsRequest?
-    return writeResponse(sessionContext, soap.response(null), true)
-
-  if acsRequest.name is 'Download'
-    if not acsRequest.url?
-      FS_PORT = config.get('FS_PORT')
-      FS_HOSTNAME = config.get('FS_HOSTNAME')
-      FS_SSL = config.get('FS_SSL')
-      acsRequest.url = if FS_SSL then 'https://' else 'http://'
-      acsRequest.url += FS_HOSTNAME
-      acsRequest.url += ":#{FS_PORT}" if FS_PORT != 80
-      acsRequest.url += "/#{encodeURI(acsRequest.fileName)}"
-
-    if not acsRequest.fileSize?
-      return localCache.getFiles((err, hash, files) ->
-        return throwError(err, sessionContext.httpResponse) if err
-
-        if acsRequest.fileName of files
-          acsRequest.fileSize = files[acsRequest.fileName].length
-        else
-          acsRequest.fileSize = 0
-
-        return sendAcsRequest(sessionContext, id, acsRequest)
-      )
-
-  rpc = {
-    id : id,
-    acsRequest : acsRequest,
-    cwmpVersion : sessionContext.cwmpVersion
-  }
-
-  logger.accessInfo({
-    sessionContext: sessionContext
-    message: 'ACS request'
-    rpc: rpc
-  })
-
-  res = soap.response(rpc)
-
-  writeResponse(sessionContext, res)
-
-
-getSession = (connection, sessionId, callback) ->
-  sessionContext = currentSessions.get(connection)
-
-  if sessionContext
-    currentSessions.delete(connection)
-    return callback(null, sessionContext)
-
-  return callback() if not sessionId
-
-  setTimeout(() ->
-    cache.pop("session_#{sessionId}", (err, sessionContextString) ->
-      return callback(err) if err or not sessionContextString
-      session.deserialize(sessionContextString, callback)
-    )
-  , 100)
-
-
-currentSessions = new WeakMap()
-
-# When socket closes, store active sessions in cache
-onConnection = (socket) ->
-  # The property remoteAddress may be undefined after the connection is
-  # closed, unless we read it at least once (caching?)
-  socket.remoteAddress
-
-  socket.on('close', () ->
-    sessionContext = currentSessions.get(socket)
-    return if not sessionContext
-    currentSessions.delete(socket)
-    now = Date.now()
-
-    lastActivity = sessionContext.lastActivity
-    timeoutMsg = logger.flatten({
-      sessionContext: sessionContext
-      message: 'Session timeout'
-      sessionTimestamp: sessionContext.timestamp
-    })
-
-    timeout = (sessionContext.lastActivity + sessionContext.timeout * 1000) - now
-    if timeout <= 0
-      return logger.accessError(timeoutMsg)
-
-    setTimeout(() ->
-      cache.get("session_#{sessionContext.sessionId}", (err, sessionContextString) ->
-        return throwError(err) if err
-        return if not sessionContextString
-        session.deserialize(sessionContextString, (err, sessionContext) ->
-          return throwError(err) if err
-          if sessionContext.lastActivity == lastActivity
-            logger.accessError(timeoutMsg)
-        )
-      )
-    , timeout + 1000).unref()
-
-    session.serialize(sessionContext, (err, sessionContextString) ->
-      return throwError(err) if err
-      cache.set("session_#{sessionContext.sessionId}", sessionContextString, Math.ceil(timeout / 1000) + 3, (err) ->
-        return throwError(err) if err
-      )
-    )
-  )
-
-setInterval(() ->
-  if stats.droppedRequests
-    logger.warn({
-      message: 'Worker overloaded'
-      droppedRequests: stats.droppedRequests
-      totalRequests: stats.totalRequests
-      initiatedSessions: stats.initiatedSessions
-      pid: process.pid
-    })
-
-  stats.totalRequests = 0
-  stats.droppedRequests = 0
-  stats.initiatedSessions = 0
-, 10000).unref()
-
-
-getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) ->
-  cache.get("#{deviceId}_tasks_faults_operations", (err, res) ->
-    return callback(err) if err
-
-    if res
-      res = JSON.parse(res)
-      return callback(null, res.tasks or [], res.faults or {}, res.operations or {})
-
-    faults = tasks = operations = cacheUntil = null
-
-    db.getFaults(deviceId, (err, _faults) ->
-      if err
-        callback?(err)
-        return callback = null
-
-      faults = _faults
-      if tasks? and operations?
-        return callback(null, tasks, faults, operations, cacheUntil)
-    )
-
-    db.getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) ->
-      if err
-        callback?(err)
-        return callback = null
-
-      tasks = dueTasks
-      cacheUntil = nextTimestamp or 0
-
-      if faults? and operations?
-        return callback(null, tasks, faults, operations, cacheUntil)
-    )
-
-    db.getOperations(deviceId, (err, _operations) ->
-      if err
-        callback?(err)
-        return callback = null
-
-      operations = _operations
-
-      if faults? and tasks?
-        return callback(null, tasks, faults, operations, cacheUntil)
-    )
-
-    if faults? and tasks? and operations?
-      return callback(null, tasks, faults, operations, cacheUntil)
-  )
-
-
-cacheDueTasksAndFaultsAndOperations = (deviceId, tasks, faults, operations, cacheUntil, callback) ->
-  v = {}
-  v.tasks = tasks if tasks.length
-  v.faults = faults if Object.keys(faults).length
-  v.operations = operations if Object.keys(operations).length
-
-  if cacheUntil
-    ttl = Math.trunc((Date.now() - cacheUntil) / 1000)
-  else
-    ttl = config.get('MAX_CACHE_TTL', deviceId)
-
-  cache.set("#{deviceId}_tasks_faults_operations", JSON.stringify(v), ttl, callback)
-
-
-processRequest = (sessionContext, rpc) ->
-  if rpc.cpeRequest?
-    if rpc.cpeRequest.name is 'Inform'
-      logger.accessInfo({
-        sessionContext: sessionContext
-        message: 'Inform'
-        rpc: rpc
-      })
-      inform(sessionContext, rpc)
-    else if rpc.cpeRequest.name is 'TransferComplete'
-      logger.accessInfo({
-        sessionContext: sessionContext
-        message: 'CPE request'
-        rpc: rpc
-      })
-      transferComplete(sessionContext, rpc)
-    else if rpc.cpeRequest.name is 'GetRPCMethods'
-      logger.accessInfo({
-        sessionContext: sessionContext
-        message: 'CPE request'
-        rpc: rpc
-      })
-      res = soap.response({
-        id : rpc.id
-        acsResponse : {name: 'GetRPCMethodsResponse', methodList: ['Inform', 'GetRPCMethods', 'TransferComplete']}
-        cwmpVersion : sessionContext.cwmpVersion
-      })
-      writeResponse(sessionContext, res)
-    else
-      return throwError(new Error('ACS method not supported'), sessionContext.httpResponse)
-  else if rpc.cpeResponse
-    session.rpcResponse(sessionContext, rpc.id, rpc.cpeResponse, (err) ->
-      return throwError(err, sessionContext.httpResponse) if err
-      nextRpc(sessionContext)
-    )
-  else if rpc.cpeFault
-    logger.accessWarn({
-      sessionContext: sessionContext
-      message: 'CPE fault'
-      rpc:rpc
-    })
-    session.rpcFault(sessionContext, rpc.id, rpc.cpeFault, (err, fault) ->
-      return throwError(err, sessionContext.httpResponse) if err
-
-      if fault
-        recordFault(sessionContext, fault)
-        session.clearProvisions(sessionContext)
-      nextRpc(sessionContext)
-    )
-  else # CPE sent empty response
-    session.timeoutOperations(sessionContext, (err, faults, operations) ->
-      return throwError(err, sessionContext.httpResponse) if err
-
-      for fault, i in faults
-        for k, v of operations[i].retries
-          sessionContext.retries[k] = v
-
-        recordFault(sessionContext, fault, operations[i].provisions, operations[i].channels)
-
-      nextRpc(sessionContext)
-    )
-
-
-listener = (httpRequest, httpResponse) ->
-  stats.totalRequests += 1
-  if httpRequest.method != 'POST'
-    httpResponse.writeHead 405, {'Allow': 'POST', 'Connection': 'close'}
-    httpResponse.end('405 Method Not Allowed')
-    return
-
-  sessionId = null
-  # Separation by comma is important as some devices don't comform to standard
-  COOKIE_REGEX = /\s*([a-zA-Z0-9\-_]+?)\s*=\s*"?([a-zA-Z0-9\-_]*?)"?\s*(,|;|$)/g
-  while match = COOKIE_REGEX.exec(httpRequest.headers.cookie)
-    sessionId = match[2] if match[1] == 'session'
-
-  # If overloaded, ask CPE to retry in 60 seconds
-  if not sessionId and stats.concurrentRequests > MAX_CONCURRENT_REQUESTS
-    httpResponse.writeHead(503, {'Retry-after': 60, 'Connection': 'close'})
-    httpResponse.end('503 Service Unavailable')
-    stats.droppedRequests += 1
-    return
-
-  if httpRequest.headers['content-encoding']?
-    switch httpRequest.headers['content-encoding']
-      when 'gzip'
-        stream = httpRequest.pipe(zlib.createGunzip())
-      when 'deflate'
-        stream = httpRequest.pipe(zlib.createInflate())
-      else
-        httpResponse.writeHead(415, {'Connection': 'close'})
-        httpResponse.end('415 Unsupported Media Type')
-        return
-  else
-    stream = httpRequest
-
-  stats.concurrentRequests += 1
-  httpRequest.on('aborted', () ->
-    stats.concurrentRequests -= 1
-    # In some cases event end can be emitted after aborted event
-    httpRequest.removeAllListeners('end')
-  )
-
-  chunks = []
-  bytes = 0
-
-  stream.on('data', (chunk) ->
-    chunks.push(chunk)
-    bytes += chunk.length
-  )
-
-  stream.on('end', () ->
-    body = new Buffer(bytes)
-    offset = 0
-    chunks.forEach((chunk) ->
-      chunk.copy(body, offset, 0, chunk.length)
-      offset += chunk.length
-    )
-
-    parsedRpc = (sessionContext, rpc, parseWarnings) ->
-      for w in parseWarnings
-        w.sessionContext = sessionContext
-        w.rpc = rpc;
-        logger.accessWarn(w)
-
-      if config.get('DEBUG', sessionContext.deviceId)
-        dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{body}\n\n"
-        require('fs').appendFile("./debug/#{sessionContext.deviceId}.dump", dump, (err) ->
-          return throwError(err) if err
-        )
-
-      processRequest(sessionContext, rpc)
-
-    getSession(httpRequest.connection, sessionId, (err, sessionContext) ->
-      return throwError(err, httpResponse) if err
-
-      if sessionContext
-        sessionContext.httpRequest = httpRequest
-        sessionContext.httpResponse = httpResponse
-
-        if sessionContext.sessionId != sessionId or
-            sessionContext.lastActivity + sessionContext.timeout * 1000 < Date.now()
-          logger.accessError({message: 'Invalid session', sessionContext: sessionContext})
-          httpResponse.writeHead(400, {'Connection': 'close'})
-          httpResponse.end('Invalid session')
-          stats.concurrentRequests -= 1
-          return
-      else if stats.concurrentRequests > MAX_CONCURRENT_REQUESTS
-        # Check again just in case device included old session ID from the previous session
-        httpResponse.writeHead(503, {'Retry-after': 60, 'Connection': 'close'})
-        httpResponse.end('503 Service Unavailable')
-        stats.droppedRequests += 1
-        stats.concurrentRequests -= 1
-        return
-
-      parseWarnings = []
-      try
-        rpc = soap.request(body, sessionContext?.cwmpVersion, parseWarnings)
-      catch err
-        logger.accessError({
-          message: 'XML parse error'
-          parseError: err.message.trim()
-          sessionContext: sessionContext or {httpRequest: httpRequest, httpResponse: httpResponse}
-        })
-        httpResponse.writeHead(400, {'Connection': 'close'})
-        httpResponse.end(err.message)
-        stats.concurrentRequests -= 1
-        return
-
-      if sessionContext
-        if rpc.cpeRequest?.name is 'Inform' or
-            (!sessionContext.rpcRequest ^ !(rpc.cpeResponse or rpc.cpeFault))
-          logger.accessError({
-            message: 'Bad session state'
-            sessionContext: sessionContext
-          })
-          httpResponse.writeHead(400, {'Connection': 'close'})
-          httpResponse.end('Bad session state')
-          stats.concurrentRequests -= 1
-          return
-
-        return parsedRpc(sessionContext, rpc, parseWarnings)
-
-      if rpc.cpeRequest?.name isnt 'Inform'
-        logger.accessError({
-          message: 'Invalid session'
-          sessionContext: sessionContext or {httpRequest: httpRequest, httpResponse: httpResponse}
-        })
-        httpResponse.writeHead(400, {'Connection': 'close'})
-        httpResponse.end('Invalid session')
-        stats.concurrentRequests -= 1
-        return
-
-      stats.initiatedSessions += 1
-      deviceId = common.generateDeviceId(rpc.cpeRequest.deviceId)
-      return session.init(deviceId, rpc.cwmpVersion, rpc.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId), (err, sessionContext) ->
-        return throwError(err, httpResponse) if err
-
-        sessionContext.httpRequest = httpRequest
-        sessionContext.httpResponse = httpResponse
-        sessionContext.sessionId = crypto.randomBytes(8).toString('hex')
-        httpRequest.connection.setTimeout(sessionContext.timeout * 1000)
-
-        getDueTasksAndFaultsAndOperations(deviceId, sessionContext.timestamp, (err, dueTasks, faults, operations, cacheUntil) ->
-          return throwError(err, httpResponse) if err
-
-          sessionContext.tasks = dueTasks
-          sessionContext.faults = faults
-          sessionContext.retries = {}
-          for k, v of faults
-            sessionContext.retries[k] = v.retries
-          sessionContext.operations = operations
-          sessionContext.cacheUntil = cacheUntil
-
-          # Delete expired faults
-          for k, v of sessionContext.faults
-            if v.expiry >= sessionContext.timestamp
-              delete sessionContext.faults[k]
-              sessionContext.faultsTouched ?= {}
-              sessionContext.faultsTouched[k] = true
-
-          parsedRpc(sessionContext, rpc, parseWarnings)
-        )
-      )
-    )
-  )
-
-
-exports.listener = listener
-exports.onConnection = onConnection