Class: SolidQueue::ClaimedExecution

Inherits:
Execution show all
Defined in:
lib/solid_queue_mongoid/models/classes.rb,
lib/solid_queue_mongoid/models/claimed_execution.rb

Defined Under Namespace

Classes: Result

Constant Summary

Constants inherited from Record

Record::INDEX_HINTS

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Execution

create_all_from_jobs, type, #type

Methods inherited from Record

create_or_find_by!, find_by!, index, index_specifications, index_specifications=, inherited, non_blocking_lock, supports_insert_conflict_target?, transaction, use_index

Class Method Details

.claiming(job_ids, process_id, &block) ⇒ Object

Atomically creates ClaimedExecution records for the given job_ids and yields the claimed set to the block (which deletes the ReadyExecutions).



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 28

def claiming(job_ids, process_id, &block)
  job_data = Array(job_ids).map { |job_id| { job_id: job_id, process_id: process_id } }

  SolidQueue.instrument(:claim, process_id: process_id, job_ids: job_ids) do |payload|
    claimed = job_data.filter_map do |attrs|
      create!(attrs)
    rescue Mongoid::Errors::Validations, Mongo::Error::OperationFailure
      nil
    end

    block.call(claimed)

    payload[:size] = claimed.size
    payload[:claimed_job_ids] = claimed.map(&:job_id)
  end
end

.discard_all_from_jobs ⇒ Object

Raises:



76
77
78
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 76

def discard_all_from_jobs(*)
  raise Execution::UndiscardableError, "Can't discard jobs in progress"
end

.discard_all_in_batches ⇒ Object

Raises:



72
73
74
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 72

def discard_all_in_batches(*)
  raise Execution::UndiscardableError, "Can't discard jobs in progress"
end

.fail_all_with(error) ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 57

def fail_all_with(error)
  executions = includes(:job).to_a
  return if executions.empty?

  SolidQueue.instrument(:fail_many_claimed) do |payload|
    executions.each do |execution|
      execution.failed_with(error)
      execution.unblock_next_job
    end
    payload[:process_ids] = executions.map(&:process_id).uniq
    payload[:job_ids] = executions.map(&:job_id).uniq
    payload[:size] = executions.size
  end
end

.release_all ⇒ Object 



45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 45

def release_all
  SolidQueue.instrument(:release_many_claimed) do |payload|
    executions = all.to_a
    executions.each do |execution|
      execution.release
    rescue Mongoid::Errors::Validations, Mongo::Error::OperationFailure
      # If ReadyExecution already exists, that's fine
    end
    payload[:size] = executions.size
  end
end

Instance Method Details

#discard ⇒ Object

Raises:



103
104
105
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 103

def discard
  raise Execution::UndiscardableError, "Can't discard a job in progress"
end

#failed_with(error) ⇒ Object 



107
108
109
110
111
112
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 107

def failed_with(error)
  Mongoid.transaction do
    job.failed_with(error)
    destroy!
  end
end

#perform ⇒ Object

Called by Pool thread — executes the job and marks it finished or failed.



82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 82

def perform
  result = execute

  if result.success?
    finished
  else
    failed_with(result.error)
    raise result.error
  end
ensure
  unblock_next_job
end

#release ⇒ Object

Release this execution back to ready (called by process deregister / prune).



96
97
98
99
100
101
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 96

def release
  SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do
    job.dispatch_bypassing_concurrency_limits
    destroy!
  end
end

#unblock_next_job ⇒ Object 



114
115
116
 
# File 'lib/solid_queue_mongoid/models/claimed_execution.rb', line 114

def unblock_next_job
  job.unblock_next_blocked_job
end