Class: GoodJob::BatchRecord

Inherits:
BaseRecord
  • Object
show all
Includes:
AdvisoryLockable, Filterable
Defined in:
app/models/good_job/batch_record.rb

Defined Under Namespace

Classes: PropertySerializer

Constant Summary

Constants included from AdvisoryLockable

AdvisoryLockable::RecordAlreadyAdvisoryLockedError

Class Method Summary collapse

Instance Method Summary collapse

Methods included from AdvisoryLockable

#advisory_lock, #advisory_lock!, #advisory_locked?, #advisory_unlock, #advisory_unlock!, #advisory_unlocked?, #lockable_column_key, #lockable_key, #owns_advisory_lock?, #with_advisory_lock

Methods inherited from BaseRecord

bind_value, migrated?, migration_pending_warning!, with_logger_silenced

Class Method Details

.jobs_finished_at_migrated? ⇒ Boolean

Returns:

  • (Boolean) 


28
29
30
 
# File 'app/models/good_job/batch_record.rb', line 28

def self.jobs_finished_at_migrated?
  column_names.include?('jobs_finished_at')
end

Instance Method Details

#_continue_discard_or_finish(job = nil, lock: true) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'app/models/good_job/batch_record.rb', line 58

def _continue_discard_or_finish(job = nil, lock: true)
  job_discarded = job && job.finished_at.present? && job.error.present?
  buffer = GoodJob::Adapter::InlineBuffer.capture do
    advisory_lock_maybe(lock) do
      Batch.within_thread(batch_id: nil, batch_callback_id: id) do
        reload

        if job_discarded && !discarded_at
          update(discarded_at: Time.current)
          on_discard.constantize.set(priority: callback_priority, queue: callback_queue_name).perform_later(to_batch, { event: :discard }) if on_discard.present?
        end

        if enqueued_at && !(self.class.jobs_finished_at_migrated? ? jobs_finished_at : finished_at) && jobs.where(finished_at: nil).count.zero?
          self.class.jobs_finished_at_migrated? ? update(jobs_finished_at: Time.current) : update(finished_at: Time.current)

          on_success.constantize.set(priority: callback_priority, queue: callback_queue_name).perform_later(to_batch, { event: :success }) if !discarded_at && on_success.present?
          on_finish.constantize.set(priority: callback_priority, queue: callback_queue_name).perform_later(to_batch, { event: :finish }) if on_finish.present?
        end

        update(finished_at: Time.current) if !finished_at && self.class.jobs_finished_at_migrated? && jobs_finished? && callback_jobs.where(finished_at: nil).count.zero?
      end
    end
  end
  buffer.call
end

#display_attributes ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'app/models/good_job/batch_record.rb', line 42

def display_attributes
  display_properties = begin
    serialized_properties
  rescue ActiveJob::DeserializationError
    JSON.parse(read_attribute_before_type_cast(:serialized_properties))
  end

  attribute_names.to_h do |name|
    if name == "serialized_properties"
      ["properties", display_properties]
    else
      [name, self[name]]
    end
  end
end

#jobs_finished? ⇒ Boolean

Returns:

  • (Boolean) 


107
108
109
 
# File 'app/models/good_job/batch_record.rb', line 107

def jobs_finished?
  self.class.jobs_finished_at_migrated? ? jobs_finished_at : finished_at
end

#jobs_finished_at ⇒ Object 



111
112
113
 
# File 'app/models/good_job/batch_record.rb', line 111

def jobs_finished_at
  self.class.jobs_finished_at_migrated? ? self[:jobs_finished_at] : self[:finished_at]
end

#properties=(value) ⇒ Object

Raises:

  • (ArgumentError) 


101
102
103
104
105
 
# File 'app/models/good_job/batch_record.rb', line 101

def properties=(value)
  raise ArgumentError, "Properties must be a Hash" unless value.is_a?(Hash)

  self.serialized_properties = value
end

#succeeded? ⇒ Boolean

Whether the batch has finished and no jobs were discarded

Returns:

  • (Boolean) 


34
35
36
 
# File 'app/models/good_job/batch_record.rb', line 34

def succeeded?
  !discarded? && finished?
end

#to_batch ⇒ Object 



38
39
40
 
# File 'app/models/good_job/batch_record.rb', line 38

def to_batch
  Batch.new(_record: self)
end