# frozen_string_literal: true
module ActiveJob
# = Active Job \Core
#
# Provides general behavior that will be included into every Active Job
# object that inherits from ActiveJob::Base.
module Core
extend ActiveSupport::Concern
# Job arguments
attr_accessor :arguments
attr_writer :serialized_arguments
# Timestamp when the job should be performed
attr_accessor :scheduled_at
# Job Identifier
attr_accessor :job_id
# Queue in which the job will reside.
attr_writer :queue_name
# Priority that the job will have (lower is more priority).
attr_writer :priority
# ID optionally provided by adapter
attr_accessor :provider_job_id
# Number of times this job has been executed (which increments on every retry, like after an exception).
attr_accessor :executions
# Hash that contains the number of times this job handled errors for each specific retry_on declaration.
# Keys are the string representation of the exceptions listed in the retry_on declaration,
# while its associated value holds the number of executions where the corresponding retry_on
# declaration handled one of its listed exceptions.
attr_accessor :exception_executions
# I18n.locale to be used during the job.
attr_accessor :locale
# Timezone to be used during the job.
attr_accessor :timezone
# Track when a job was enqueued
attr_accessor :enqueued_at
# Track whether the adapter received the job successfully.
attr_writer :successfully_enqueued # :nodoc:
def successfully_enqueued?
@successfully_enqueued
end
# Track any exceptions raised by the backend so callers can inspect the errors.
attr_accessor :enqueue_error
# These methods will be included into any Active Job object, adding
# helpers for de/serialization and creation of job instances.
module ClassMethods
# Creates a new job instance from a hash created with +serialize+
def deserialize(job_data)
job = job_data["job_class"].constantize.new
job.deserialize(job_data)
job
end
# Creates a job preconfigured with the given options. You can call
# perform_later with the job arguments to enqueue the job with the
# preconfigured options
#
# ==== Options
# * :wait - Enqueues the job with the specified delay
# * :wait_until - Enqueues the job at the time specified
# * :queue - Enqueues the job on the specified queue
# * :priority - Enqueues the job with the specified priority
#
# ==== Examples
#
# VideoJob.set(queue: :some_queue).perform_later(Video.last)
# VideoJob.set(wait: 5.minutes).perform_later(Video.last)
# VideoJob.set(wait_until: Time.now.tomorrow).perform_later(Video.last)
# VideoJob.set(queue: :some_queue, wait: 5.minutes).perform_later(Video.last)
# VideoJob.set(queue: :some_queue, wait_until: Time.now.tomorrow).perform_later(Video.last)
# VideoJob.set(queue: :some_queue, wait: 5.minutes, priority: 10).perform_later(Video.last)
def set(options = {})
ConfiguredJob.new(self, options)
end
end
# Creates a new job instance. Takes the arguments that will be
# passed to the perform method.
def initialize(*arguments)
@arguments = arguments
@job_id = SecureRandom.uuid
@queue_name = self.class.queue_name
@priority = self.class.priority
@executions = 0
@exception_executions = {}
@timezone = Time.zone&.name
end
ruby2_keywords(:initialize)
# Returns a hash with the job data that can safely be passed to the
# queuing adapter.
def serialize
{
"job_class" => self.class.name,
"job_id" => job_id,
"provider_job_id" => provider_job_id,
"queue_name" => queue_name,
"priority" => priority,
"arguments" => serialize_arguments_if_needed(arguments),
"executions" => executions,
"exception_executions" => exception_executions,
"locale" => I18n.locale.to_s,
"timezone" => timezone,
"enqueued_at" => Time.now.utc.iso8601(9)
}
end
# Attaches the stored job data to the current instance. Receives a hash
# returned from +serialize+
#
# ==== Examples
#
# class DeliverWebhookJob < ActiveJob::Base
# attr_writer :attempt_number
#
# def attempt_number
# @attempt_number ||= 0
# end
#
# def serialize
# super.merge('attempt_number' => attempt_number + 1)
# end
#
# def deserialize(job_data)
# super
# self.attempt_number = job_data['attempt_number']
# end
#
# rescue_from(Timeout::Error) do |exception|
# raise exception if attempt_number > 5
# retry_job(wait: 10)
# end
# end
def deserialize(job_data)
self.job_id = job_data["job_id"]
self.provider_job_id = job_data["provider_job_id"]
self.queue_name = job_data["queue_name"]
self.priority = job_data["priority"]
self.serialized_arguments = job_data["arguments"]
self.executions = job_data["executions"]
self.exception_executions = job_data["exception_executions"]
self.locale = job_data["locale"] || I18n.locale.to_s
self.timezone = job_data["timezone"] || Time.zone&.name
self.enqueued_at = job_data["enqueued_at"]
end
# Configures the job with the given options.
def set(options = {}) # :nodoc:
self.scheduled_at = options[:wait].seconds.from_now.to_f if options[:wait]
self.scheduled_at = options[:wait_until].to_f if options[:wait_until]
self.queue_name = self.class.queue_name_from_part(options[:queue]) if options[:queue]
self.priority = options[:priority].to_i if options[:priority]
self
end
private
def serialize_arguments_if_needed(arguments)
if arguments_serialized?
@serialized_arguments
else
serialize_arguments(arguments)
end
end
def deserialize_arguments_if_needed
if arguments_serialized?
@arguments = deserialize_arguments(@serialized_arguments)
@serialized_arguments = nil
end
end
def serialize_arguments(arguments)
Arguments.serialize(arguments)
end
def deserialize_arguments(serialized_args)
Arguments.deserialize(serialized_args)
end
def arguments_serialized?
defined?(@serialized_arguments) && @serialized_arguments
end
end
end