Class: SolidQueue::Batch

Inherits:
Record
  • Object
show all
Includes:
Callbacks, Clearable, Status, Sweepable
Defined in:
app/models/solid_queue/batch.rb,
app/models/solid_queue/batch/status.rb,
app/models/solid_queue/batch/callbacks.rb,
app/models/solid_queue/batch/clearable.rb,
app/models/solid_queue/batch/sweepable.rb

Defined Under Namespace

Modules: Callbacks, Clearable, Status, Sweepable Classes: AlreadyFinished, PendingMigrations

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Status

#completed_jobs, #enqueued?, #failed?, #failed_jobs, #finished?, #pending_jobs, #progress_percentage, #status, #succeeded?

Methods inherited from Record

non_blocking_lock, supports_insert_conflict_target?, use_index, warn_about_pending_migrations

Class Method Details

.current_batch_id ⇒ Object 



46
47
48
 
# File 'app/models/solid_queue/batch.rb', line 46

def current_batch_id
  ActiveSupport::IsolatedExecutionState[:current_batch_id]
end

.enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, metadata: nil, **extra_metadata, &block) ⇒ Object

Raises:



37
38
39
40
41
42
43
44
 
# File 'app/models/solid_queue/batch.rb', line 37

def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, metadata: nil, **extra_metadata, &block)
  raise PendingMigrations unless migrated?

  new.tap do |batch|
    batch.assign_attributes(description:, on_success:, on_failure:, on_finish:, metadata: (metadata || {}).merge(extra_metadata))
    batch.enqueue(&block)
  end
end

.migrated? ⇒ Boolean

The batches schema ships as an optional migration in Solid Queue 1.x and becomes part of the base schema in 2.0. Until the app has run the migration, jobs enqueue without any batch bookkeeping and batches themselves can't be used.

Returns:

  • (Boolean) 


33
34
35
 
# File 'app/models/solid_queue/batch.rb', line 33

def migrated?
  @migrated ||= table_exists? && BatchExecution.table_exists? && Job.column_names.include?("batch_id")
end

.wrap_in_batch_context(batch_id) ⇒ Object 



50
51
52
53
54
55
56
 
# File 'app/models/solid_queue/batch.rb', line 50

def wrap_in_batch_context(batch_id)
  previous_batch_id = current_batch_id.presence
  ActiveSupport::IsolatedExecutionState[:current_batch_id] = batch_id
  yield
ensure
  ActiveSupport::IsolatedExecutionState[:current_batch_id] = previous_batch_id
end

Instance Method Details

#enqueue(&block) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'app/models/solid_queue/batch.rb', line 59

def enqueue(&block)
  # Fast-fail for the common case. create_all_from_jobs atomically guards
  # concurrent additions when it creates their tracking rows.
  if finished?
    raise AlreadyFinished, "Can't enqueue an already finished batch"
  end

  transaction do
    save! if new_record?

    self.class.wrap_in_batch_context(id) { block&.call(self) }

    if ActiveRecord.respond_to?(:after_all_transactions_commit)
      ActiveRecord.after_all_transactions_commit { start }
    end
  end
end

#finish ⇒ Object 



90
91
92
93
94
95
96
97
98
 
# File 'app/models/solid_queue/batch.rb', line 90

def finish
  return if finished? || !enqueued?
  return if batch_executions.exists?

  transaction do
    updated = Batch.where(id: id).unfinished.enqueued.without_executions.update_all(finished_at: Time.current)
    finalize if updated > 0
  end
end

#metadata ⇒ Object 



77
78
79
 
# File 'app/models/solid_queue/batch.rb', line 77

def metadata
  (super || {}).with_indifferent_access
end

#start ⇒ Object 



81
82
83
84
85
86
87
88
 
# File 'app/models/solid_queue/batch.rb', line 81

def start
  mark_as_enqueued

  # Refresh enqueued_at after marking as enqueued, and let a batch that started
  # with no jobs finish right away
  reload
  finish
end