<commit_before>require 'date'
require 'disque'
require 'msgpack'

require_relative 'disc/version'

class Disc
  attr_reader :disque,
    :disque_timeout

  def self.disque
    @disque ||= Disque.new(
      ENV.fetch('DISQUE_NODES', 'localhost:7711'),
      auth: ENV.fetch('DISQUE_AUTH', nil),
      cycle: Integer(ENV.fetch('DISQUE_CYCLE', '1000'))
    )
  end

  def self.disque=(disque)
    @disque = disque
  end

  def self.disque_timeout
    @disque_timeout ||= 100
  end

  def self.disque_timeout=(timeout)
    @disque_timeout = timeout
  end

  def self.default_queue
    @default_queue ||= 'default'
  end

  def self.default_queue=(queue)
    @default_queue = queue
  end

  def self.on_error(exception, job)
    $stderr.puts exception
  end

  class Worker
    attr_reader :disque,
                :queues,
                :timeout,
                :count

    def self.run
      new.run
    end

    def initialize(options = {})
      @disque = options.fetch(:disque, Disc.disque)
      @queues = options.fetch(
        :queues,
        ENV.fetch('QUEUES', Disc.default_queue)
      ).split(',')
      @count = Integer(
        options.fetch(
          :count,
          ENV.fetch('DISQUE_COUNT', '1')
        )
      )
      @timeout = Integer(
        options.fetch(
          :timeout,
          ENV.fetch('DISQUE_TIMEOUT', '2000')
        )
      )

      self.run if options[:run]
      self
    end

    def run
      $stdout.puts("Disc::Worker listening in #{queues}")
      loop do
        jobs = disque.fetch(from: queues, timeout: timeout, count: count)
        Array(jobs).each do |queue, msgid, serialized_job|
          begin
            job = MessagePack.unpack(serialized_job)
            instance = Object.const_get(job['class']).new
            instance.perform(*job['arguments'])
            disque.call('ACKJOB', msgid)
          rescue => err
            Disc.on_error(err, job.update('id' => msgid, 'queue' => queue))
          end
        end
      end
    end
  end

  module Job
    attr_reader :arguments,
                :disque,
                :disc_options

    def self.included(base)
      base.extend(ClassMethods)
    end

    module ClassMethods
      def disque
        defined?(@disque) ? @disque : Disc.disque
      end

      def disque=(disque)
        @disque = disque
      end

      def disc(queue: nil, **options)
        @queue = queue
        @disc_options = options
      end

      def disc_options
        @disc_options ||= {}
      end<commit_msg>Quit connection after fetch.<commit_after>require 'date'
require 'disque'
require 'msgpack'

require_relative 'disc/version'

class Disc
  attr_reader :disque,
    :disque_timeout

  def self.disque
    @disque ||= Disque.new(
      ENV.fetch('DISQUE_NODES', 'localhost:7711'),
      auth: ENV.fetch('DISQUE_AUTH', nil),
      cycle: Integer(ENV.fetch('DISQUE_CYCLE', '1000'))
    )
  end

  def self.disque=(disque)
    @disque = disque
  end

  def self.disque_timeout
    @disque_timeout ||= 100
  end

  def self.disque_timeout=(timeout)
    @disque_timeout = timeout
  end

  def self.default_queue
    @default_queue ||= 'default'
  end

  def self.default_queue=(queue)
    @default_queue = queue
  end

  def self.on_error(exception, job)
    $stderr.puts exception
  end

  class Worker
    attr_reader :disque,
                :queues,
                :timeout,
                :count

    def self.run
      new.run
    end

    def initialize(options = {})
      @disque = options.fetch(:disque, Disc.disque)
      @queues = options.fetch(
        :queues,
        ENV.fetch('QUEUES', Disc.default_queue)
      ).split(',')
      @count = Integer(
        options.fetch(
          :count,
          ENV.fetch('DISQUE_COUNT', '1')
        )
      )
      @timeout = Integer(
        options.fetch(
          :timeout,
          ENV.fetch('DISQUE_TIMEOUT', '2000')
        )
      )

      self.run if options[:run]
      self
    end

    def run
      $stdout.puts("Disc::Worker listening in #{queues}")
      loop do
        jobs = disque.fetch(from: queues, timeout: timeout, count: count)
        Array(jobs).each do |queue, msgid, serialized_job|
          begin
            job = MessagePack.unpack(serialized_job)
            instance = Object.const_get(job['class']).new
            instance.perform(*job['arguments'])
            disque.call('ACKJOB', msgid)
          rescue => err
            Disc.on_error(err, job.update('id' => msgid, 'queue' => queue))
          end
        end
      end
    ensure
      disque.quit
    end
  end

  module Job
    attr_reader :arguments,
                :disque,
                :disc_options

    def self.included(base)
      base.extend(ClassMethods)
    end

    module ClassMethods
      def disque
        defined?(@disque) ? @disque : Disc.disque
      end

      def disque=(disque)
        @disque = disque
      end

      def disc(queue: nil, **options)
        @queue = queue
        @disc_options = options
      end

      def disc_options
        @disc_options ||= {}
      end
{'fraction_non_alphanumeric': 0.11097946287519747, 'fraction_numerical': 0.00631911532385466, 'mean_word_length': 1.9550758459743292, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '444757', 'n_tokens_mistral': 1661, 'n_tokens_neox': 1573, 'n_words': 348}