<commit_before>require 'digest'

module Bellbro
  class Service < Bellbro::Bell
    include Bellbro::SidekiqUtils
    include Bellbro::Trackable

    SLEEP_INTERVAL = defined?(Rails) && Rails.env.test? ? 1 : 3600
    track_with_schema jobs_started: Integer

    attr_reader :thread, :thread_error, :jid

    def initialize
      @done = false
      @jid = Digest::MD5.hexdigest(Time.now.utc.to_s + Thread.current.object_id.to_s)
    end

    def start
      @thread = Thread.new do
        begin
          run
        rescue Exception => @thread_error
          ring "#{@thread_error.inspect}", type: :error
          Airbrake.notify(@thread_error)
          raise @thread_error
        end
      end
    end

    def stop
      @done = true
      ring "Stopping #{self.class} service..."
      @thread.join
      ring "#{self.class.to_s.capitalize} service stopped."
    end

    def run
      ring "Starting #{self.class} service."
      Service.mutex.synchronize { track }
      begin
        start_jobs
        Service.mutex.synchronize { status_update }
        sleep self.class::SLEEP_INTERVAL
      end until @done
      Service.mutex.synchronize { stop_tracking }
    end

    def start_jobs
      each_job do |job|
        klass = job[:klass].constantize
        jid = klass.perform_async(job[:arguments])
        ring "Starting job #{jid} #{job[:klass]} with #{job[:arguments]}."
        record_incr(:jobs_started)
      end
    end

    def each_job
      # Override
      []
    end

    def running?
      !@done
    end

    def self.mutex
      $mutex ||= Mutex.new
    end
  end

end<commit_msg>Clean up service sleep behavior<commit_after>require 'digest'

module Bellbro
  class Service < Bellbro::Bell
    include Bellbro::SidekiqUtils
    include Bellbro::Trackable

    attr_reader :thread, :thread_error, :jid

    def self.poll_interval(arg)
      @sleep_interval = arg
    end

    poll_interval defined?(Rails) && Rails.env.test? ? 1 : 3600
    track_with_schema jobs_started: Integer

    def initialize
      @done = false
      @jid = Digest::MD5.hexdigest(Time.now.utc.to_s + Thread.current.object_id.to_s)
    end

    def start
      @thread = Thread.new do
        begin
          run
        rescue Exception => @thread_error
          ring "#{@thread_error.inspect}", type: :error
          Airbrake.notify(@thread_error)
          raise @thread_error
        end
      end
    end

    def stop
      @done = true
      ring "Stopping #{self.class} service..."
      @thread.join
      ring "#{self.class.to_s.capitalize} service stopped."
    end

    def run
      ring "Starting #{self.class} service."
      Service.mutex.synchronize { track }
      begin
        start_jobs
        Service.mutex.synchronize { status_update }
        sleep
      end until @done
      Service.mutex.synchronize { stop_tracking }
    end

    def start_jobs
      each_job do |job|
        klass = job[:klass].constantize
        jid = klass.perform_async(job[:arguments])
        ring "Starting job #{jid} #{job[:klass]} with #{job[:arguments]}."
        record_incr(:jobs_started)
      end
    end

    def sleep
      super(self.class.sleep_interval)
    end

    def each_job
      # Override
      []
    end

    def running?
      !@done
    end

    def self.mutex
      $mutex ||= Mutex.new
    end

    def self.sleep_interval
      @sleep_interval
    end

  end

end
{'fraction_non_alphanumeric': 0.09497206703910614, 'fraction_numerical': 0.0035283740076448105, 'mean_word_length': 2.1383763837638377, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '7211338', 'n_tokens_mistral': 1093, 'n_tokens_neox': 1046, 'n_words': 253}