<commit_before>var fs = require('fs')
var path = require('path')
var net = require('net')
var winston = require('winston')
var ndjson = require('ndjson')
var makeSockPath = require('./makeSockPath')

var workingDir = process.env.NODE_TASK_CWD || __dirname
var errorLogPath = path.resolve(workingDir, './ruby_node_task-error.log')

// write pidfile
fs.writeFileSync(path.resolve(workingDir, './ruby_node_task.pid'), process.pid+'')

var log = new winston.Logger({
  transports: [
    process.env.NODE_TASK_DEBUG ? new winston.transports.File({
      filename: path.resolve(workingDir, 'ruby_node_task-debug.log'),
      level: 'debug',
    }) : new winston.transports.Console(),
  ],
})

var sockPath = process.env.NODE_TASK_SOCK_PATH || makeSockPath(
  workingDir,
  process.env.NODE_TASK_DAEMON_ID || 'ruby_node_task'
)

function serialiseMsg(msg) {
  return JSON.stringify(msg)+'\n'
}

function errorMsg(err) {
  return {error: {message: err.message, code: err.code}}
}

var server = net.createServer(function(socket) {
  log.debug('client connected')

  // use end rather than close as once ended client intends to disconnect
  socket.on('end', function() {
    log.debug('client finished')
  })

  socket.on('close', function() {
    log.debug('client disconnect')
  })

  function sendMsg(msg) { socket.end(serialiseMsg(msg)) }
  function sendError(err) { socket.end(serialiseMsg(errorMsg(err))) }

  var busy = false
  socket
    .on('error', function(err) {
      log.error('socket error: '+err.toString())
    })
    .pipe(ndjson.parse())
    .on('error', sendError)
    .on('data', function(msg) {
      if (busy) {
        return sendError(new Error('only one task can be run per connection'))
      }

      if (msg.status) {
        return server.getConnections(function(err, count) {
          sendMsg({clients: count-1}) // minus this connection
        })
      }

      if (!msg.task) {
        return sendError(new Error('msg.task not defined'))
      }

      try {
        var runTask = require(path.resolve(workingDir, msg.task))
      } catch (err) {
        return sendError(err)
      }

      var opts = msg.opts || {}
      busy = true
      runTask(opts, function(err, result) {
        busy = false
        if (err) {
          log.error('task error: '+err)
          sendError(err)
        } else {
          log.debug('task complete: '+msg.task)
          sendMsg({result: result})
        }
      })
    })
})

server.on('error', function(err) {
  log.error('server error: '+err.toString())
})

server.listen(sockPath, function() {
  log.debug('listening on '+sockPath)
})

process.on('exit', function() {
  log.debug('removing '+sockPath)
  try {
    fs.unlinkSync(sockPath)
  } catch (err) {
    // already removed
  }
})

process.on('uncaughtException', function(err) {
  log.error('uncaught exception: '+err.toString(), function () {
    try {
      fs.writeFileSync(errorLogPath, err.stack)
    } catch (err) {
      return log.error('logging error: '+err.toString(), function () {
        process.exit()
      })
    }
    process.exit()
  })
})<commit_msg>refactored event handlers for clarity<commit_after>var fs = require('fs')
var path = require('path')
var net = require('net')
var winston = require('winston')
var ndjson = require('ndjson')

var workingDir = path.resolve(process.env.NODE_TASK_CWD || __dirname)
var errorLogPath = path.join(workingDir, 'ruby_node_task-error.log')

// write pidfile
fs.writeFileSync(path.join(workingDir, 'ruby_node_task.pid'), process.pid+'')

var logger = makeLogger(workingDir, process.env.NODE_TASK_DEBUG)

var sockPath = process.env.NODE_TASK_SOCK_PATH || makeSockPath(
  workingDir,
  process.env.NODE_TASK_DAEMON_ID || 'ruby_node_task'
)

var server = net.createServer(onClientConnect)
  .on('error', function (err) { logger.error('server error: '+err.toString()) })
  .listen(sockPath, function() { logger.debug('listening on '+sockPath) })

process.on('exit', onExit)
process.on('uncaughtException', onUncaughtException)

// the important part
function onClientConnect(socket) {
  logger.debug('client connected')

  var busy = false
  socket
    .on('error', function(err) { logger.error('socket error: '+err.toString()) })
    .on('end', function() { logger.debug('client finished') })
    .on('close', function() { logger.debug('client disconnect') })
    .pipe(ndjson.parse())
    .on('error', sendError)
    .on('data', receiveMsg)
  
  function sendMsg(msg) { socket.end(serialiseMsg(msg)) }
  function sendError(err) { socket.end(serialiseMsg(errorMsg(err))) }
  function receiveMsg(msg) {
    if (busy) {
      return sendError(new Error('only one task can be run per connection'))
    }

    if (msg.status) {
      return server.getConnections(function(err, count) {
        sendMsg({clients: count-1}) // minus this connection
      })
    }

    if (!msg.task) {
      return sendError(new Error('msg.task not defined'))
    }

    var runTask = loadTaskModule(path.join(workingDir, msg.task))

    var opts = msg.opts || {}
    busy = true
    runTask(opts, function(err, result) {
      busy = false
      if (err) {
        logger.error('task error: '+err)
        sendError(err)
      } else {
        logger.debug('task complete: '+msg.task)
        sendMsg({result: result})
      }
    })
  }
  
  function loadTaskModule(taskModule) {
    try {
      return require(taskModule)
    } catch (err) {
      return sendError(new Error('Encountered "'+err+'" while attempting to load task "'+taskModule+'"'))
    }
  }
}

function onUncaughtException(err) {
  logger.error('uncaught exception: '+err.toString(), function () {
    try {
      fs.writeFileSync(errorLogPath, err.stack)
    } catch (err) {
      return logger.error('logging error: '+err.toString(), function () {
        process.exit()
      })
    }
    process.exit()
  })
}

function onExit() {
  try {
    fs.unlinkSync(sockPath)
    logger.debug('removed '+sockPath)
  } catch (err) {
    // already removed
  }
}

function makeLogger(workingDir, debug) {
  return new winston.Logger({
    transports: [
      debug ? new winston.transports.File({
        filename: path.join(workingDir, 'ruby_node_task-debug.log'),
        level: 'debug',
      }) : new winston.transports.Console(),
    ],
  })
}

// make a path for a cross platform compatible socket path
function makeSockPath(dir, name) {
  if (process.platform === 'win32') {
    return '\\\\.\\pipe\\'+name+'\\' + path.resolve(dir)
  } else {
    return path.join(dir, name+'.sock')
  }
}

function serialiseMsg(msg) {
  return JSON.stringify(msg)+'\n'
}

function errorMsg(err) {
  return {
    error: {
      message: err.message,
      code: err.code,
      stack: err.stack,
    }
  }
}
{'fraction_non_alphanumeric': 0.15334226589251154, 'fraction_numerical': 0.000595503945213637, 'mean_word_length': 3.2013758599124453, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '5072998', 'n_tokens_mistral': 2216, 'n_tokens_neox': 2168, 'n_words': 460}