<commit_before>###
# Copyright 2013-2017  <NAME>
#
# 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 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<commit_msg>Expire fault when original task expires<commit_after>###
# Copyright 2013-2017  <NAME>
#
# 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
{'fraction_non_alphanumeric': 0.1079465159256441, 'fraction_numerical': 0.005109251005544081, 'mean_word_length': 3.3998087039693927, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6294303', 'n_tokens_mistral': 2670, 'n_tokens_neox': 2582, 'n_words': 739}