<commit_before># encoding: UTF-8
require 'active_support/concern'

# Worker behavior for a job
module RocketJob
  module Plugins
    module Job
      module Worker
        extend ActiveSupport::Concern

        included do
          # Run this job later
          #
          # Saves it to the database for processing later by workers
          def self.perform_later(*args, &block)
            if RocketJob::Config.inline_mode
              perform_now(*args, &block)
            else
              job = new(arguments: args)
              block.call(job) if block
              job.save!
              job
            end
          end

          # Run this job now.
          #
          # The job is not saved to the database since it is processed entriely in memory
          # As a result before_save and before_destroy callbacks will not be called.
          # Validations are still called however prior to calling #perform
          def self.perform_now(*args, &block)
            job = new(arguments: args)
            block.call(job) if block
            job.perform_now
            job
          end

          # Returns the next job to work on in priority based order
          # Returns nil if there are currently no queued jobs, or processing batch jobs
          #   with records that require processing
          #
          # Parameters
          #   worker_name [String]
          #     Name of the worker that will be processing this job
          #
          #   skip_job_ids [Array<BSON::ObjectId>]
          #     Job ids to exclude when looking for the next job
          #
          # Note:
          #   If a job is in queued state it will be started
          def self.rocket_job_next_job(worker_name, skip_job_ids = nil)
            while (job = rocket_job_retrieve(worker_name, skip_job_ids))
              case
              when job.running?
                # Batch Job
                return job
              when job.expired?
                job.rocket_job_fail_on_exception!(worker_name) { job.destroy }
                logger.info "Destroyed expired job #{job.class.name}, id:#{job.id}"
              else
                job.worker_name = worker_name
                job.rocket_job_fail_on_exception!(worker_name) { job.start }
                return job if job.running?
              end
            end
          end

          # Requeues all jobs that were running on worker that died
          def self.requeue_dead_worker(worker_name)
            running.each { |job| job.requeue!(worker_name) if job.may_requeue?(worker_name) }
          end

          # Turn off embedded callbacks. Slow and not used for Jobs
          embedded_callbacks_off
        end

        # Runs the job now in the current thread.
        #
        # Validations are called prior to running the job.
        #
        # The job is not saved and therefore the following callbacks are _not_ called:
        # * before_save
        # * after_save
        # * before_create
        # * after_create
        #
        # Exceptions are _not_ suppressed and should be handled by the caller.
        def perform_now
          # Call validations
          if respond_to?(:validate!)
            validate!
          elsif invalid?
            raise(MongoMapper::DocumentNotValid, self)
          end
          worker = RocketJob::Worker.new(name: 'inline')
          worker.started
          start if may_start?
          # Raise exceptions
          rocket_job_work(worker, true) if running?
          result
        end

        def perform(*)
          raise NotImplementedError
        end

        # Fail this job in the event of an exception.
        #
        # The job is automatically saved only if an exception is raised in the supplied block.
        #
        # worker_name: [String]
        #   Name of the worker on which the exception has occurred
        #
        # raise_exceptions: [true|false]
        #   Re-raise the exception after updating the job
        #   Default: !RocketJob::Config.inline_mode
        def rocket_job_fail_on_exception!(worker_name, raise_exceptions = !RocketJob::Config.inline_mode)
          yield
        rescue Exception => exc
          if failed? || !may_fail?
            self.exception        = JobException.from_exception(exc)
            exception.worker_name = worker_name
          else
            fail(worker_name, exc)
          end
          save! unless new_record?
          raise exc if raise_exceptions
        end

        private

        # Works on this job
        #
        # Returns [true|false] whether this job should be excluded from the next lookup
        #
        # If an exception is thrown the job is marked as failed and the exception
        # is set in the job itself.
        #
        # Thread-safe, can be called by multiple threads at the same time<commit_msg>rocket_job_work needs to be public since it is called from the worker.<commit_after># encoding: UTF-8
require 'active_support/concern'

# Worker behavior for a job
module RocketJob
  module Plugins
    module Job
      module Worker
        extend ActiveSupport::Concern

        included do
          # Run this job later
          #
          # Saves it to the database for processing later by workers
          def self.perform_later(*args, &block)
            if RocketJob::Config.inline_mode
              perform_now(*args, &block)
            else
              job = new(arguments: args)
              block.call(job) if block
              job.save!
              job
            end
          end

          # Run this job now.
          #
          # The job is not saved to the database since it is processed entriely in memory
          # As a result before_save and before_destroy callbacks will not be called.
          # Validations are still called however prior to calling #perform
          def self.perform_now(*args, &block)
            job = new(arguments: args)
            block.call(job) if block
            job.perform_now
            job
          end

          # Returns the next job to work on in priority based order
          # Returns nil if there are currently no queued jobs, or processing batch jobs
          #   with records that require processing
          #
          # Parameters
          #   worker_name [String]
          #     Name of the worker that will be processing this job
          #
          #   skip_job_ids [Array<BSON::ObjectId>]
          #     Job ids to exclude when looking for the next job
          #
          # Note:
          #   If a job is in queued state it will be started
          def self.rocket_job_next_job(worker_name, skip_job_ids = nil)
            while (job = rocket_job_retrieve(worker_name, skip_job_ids))
              case
              when job.running?
                # Batch Job
                return job
              when job.expired?
                job.rocket_job_fail_on_exception!(worker_name) { job.destroy }
                logger.info "Destroyed expired job #{job.class.name}, id:#{job.id}"
              else
                job.worker_name = worker_name
                job.rocket_job_fail_on_exception!(worker_name) { job.start }
                return job if job.running?
              end
            end
          end

          # Requeues all jobs that were running on worker that died
          def self.requeue_dead_worker(worker_name)
            running.each { |job| job.requeue!(worker_name) if job.may_requeue?(worker_name) }
          end

          # Turn off embedded callbacks. Slow and not used for Jobs
          embedded_callbacks_off
        end

        # Runs the job now in the current thread.
        #
        # Validations are called prior to running the job.
        #
        # The job is not saved and therefore the following callbacks are _not_ called:
        # * before_save
        # * after_save
        # * before_create
        # * after_create
        #
        # Exceptions are _not_ suppressed and should be handled by the caller.
        def perform_now
          # Call validations
          if respond_to?(:validate!)
            validate!
          elsif invalid?
            raise(MongoMapper::DocumentNotValid, self)
          end
          worker = RocketJob::Worker.new(name: 'inline')
          worker.started
          start if may_start?
          # Raise exceptions
          rocket_job_work(worker, true) if running?
          result
        end

        def perform(*)
          raise NotImplementedError
        end

        # Fail this job in the event of an exception.
        #
        # The job is automatically saved only if an exception is raised in the supplied block.
        #
        # worker_name: [String]
        #   Name of the worker on which the exception has occurred
        #
        # raise_exceptions: [true|false]
        #   Re-raise the exception after updating the job
        #   Default: !RocketJob::Config.inline_mode
        def rocket_job_fail_on_exception!(worker_name, raise_exceptions = !RocketJob::Config.inline_mode)
          yield
        rescue Exception => exc
          if failed? || !may_fail?
            self.exception        = JobException.from_exception(exc)
            exception.worker_name = worker_name
          else
            fail(worker_name, exc)
          end
          save! unless new_record?
          raise exc if raise_exceptions
        end

        # Works on this job
        #
        # Returns [true|false] whether this job should be excluded from the next lookup
        #
        # If an exception is thrown the job is marked as failed and the exception
        # is set in the job itself.
        #
        # Thread-safe, can be called by multiple threads at the same time
{'fraction_non_alphanumeric': 0.055065627563576705, 'fraction_numerical': 0.00020508613617719443, 'mean_word_length': 1.6338104239805562, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '1195641', 'n_tokens_mistral': 2483, 'n_tokens_neox': 2435, 'n_words': 881}