<commit_before>_         = require 'lodash'
commander = require 'commander'
redis     = require 'redis'
RedisNS   = require '@octoblu/redis-ns'

class Command
  constructor: ->

  parseOptions: =>
    commander
      .option '-i, --input-ns <namespace>', 'Namespace from which to siphon (env: SIPHON_INPUT_NS)'
      .option '-o, --output-ns <namespace>', 'Namespace to siphon in to (env: SIPHON_OUTPUT_NS)'
      .option '-r, --rate <rate>', 'Rate at which to siphon work, defaults to x'
      .parse process.argv

    @inputNS = commander.inputQueue ? process.env.SIPHON_INPUT_NS
    @outputNS = commander.outputQueue ? process.env.SIPHON_OUTPUT_NS

    unless @inputNS && @outputNS
      commander.outputHelp()
      process.exit 1

  run: =>
    @parseOptions()

    inClient     = _.bindAll new RedisNS(@inputNS, redis.createClient @redisUri)
    inJobManager = new JobManager client: inClient

    outClient     = _.bindAll new RedisNS(@outputNS, redis.createClient @redisUri)
    outJobManager = new JobManager client: outClient

    timeoutSeconds = 1

    siphon = new Siphon {inClient, outClient, timeoutSeconds}
    siphon.do()

command = new Command
command.run()
<commit_msg>Add timeout<commit_after>_         = require 'lodash'
async     = require 'async'
commander = require 'commander'
redis     = require 'redis'
RedisNS   = require '@octoblu/redis-ns'
Siphon    = require './src/siphon'

class Command
  constructor: ->

  parseOptions: =>
    commander
      .option '-i, --input-ns <namespace>',  'Namespace from which to siphon (env: SIPHON_INPUT_NS)'
      .option '-o, --output-ns <namespace>', 'Namespace to siphon in to (env: SIPHON_OUTPUT_NS)'
      .option '-t, --timeout <seconds>', 'Timeout in seconds when waiting for jobs, default 30s (env: SIPHON_TIMEOUT)'
      .parse process.argv

    @inputNS = commander.inputNs ? process.env.SIPHON_INPUT_NS
    @outputNS = commander.outputNs ? process.env.SIPHON_OUTPUT_NS
    @timeoutSeconds = parseInt(commander.timeout ? process.env.SIPHON_OUTPUT_NS || 30)

    unless @inputNS && @outputNS
      commander.outputHelp()
      process.exit 1

  run: =>
    @parseOptions()

    inClient  = new RedisNS(@inputNS, redis.createClient @redisUri)
    inClient  = _.bindAll inClient, _.functions(inClient)
    outClient = new RedisNS(@outputNS, redis.createClient @redisUri)
    outClient = _.bindAll outClient, _.functions(outClient)

    siphon = new Siphon {inClient, outClient, @timeoutSeconds}
    async.forever siphon.do, (error) =>
      console.error error.stack
      console.error 'No longer accepting new work'

command = new Command
command.run()

{'fraction_non_alphanumeric': 0.1130567186905215, 'fraction_numerical': 0.0026646364674533687, 'mean_word_length': 3.372712146422629, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 16, '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': '4066970', 'n_tokens_mistral': 863, 'n_tokens_neox': 851, 'n_words': 223}