Class: RubyReactor::StepWorker

Inherits:
Object
  • Object
show all
Defined in:
lib/ruby_reactor/step_worker.rb

Overview

The body of one dispatched async_step, shared by every queueing backend (Adapters::Sidekiq::StepWorker, Adapters::ActiveJob::StepWorker) exactly as Map::ElementExecutor is shared by the map element workers.

It is deliberately NOT a reactor run: it loads the parent context, resolves just this one step's arguments against it, runs the step body, writes the durable Step Result Record, and publishes the completion signal. Ordering is load-bearing — the record is written BEFORE the signal, so a reader that misses the (at-most-once) signal still finds the answer on its next fallback re-check.

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(root_context_id:, reactor_class_name:, step_context_id:, step_name:) ⇒ StepWorker

Returns a new instance of StepWorker.



33
34
35
36
37
38
 
# File 'lib/ruby_reactor/step_worker.rb', line 33

def initialize(root_context_id:, reactor_class_name:, step_context_id:, step_name:)
  @root_context_id = root_context_id
  @reactor_class_name = reactor_class_name
  @step_context_id = step_context_id || root_context_id
  @step_name = step_name
end

Class Method Details

.perform(arguments) ⇒ Object 



16
17
18
19
 
# File 'lib/ruby_reactor/step_worker.rb', line 16

def perform(arguments)
  arguments = arguments.transform_keys(&:to_sym)
  new(**slice_arguments(arguments)).perform
end

Instance Method Details

#perform ⇒ Object

The lock is what makes a lost unit recoverable: the record alone cannot say whether a dispatched unit is mid-flight or gone, so StepSweeper reads this lock as the liveness signal. It also drops a duplicate delivery rather than running the body a second time.



44
45
46
47
48
49
50
51
 
# File 'lib/ruby_reactor/step_worker.rb', line 44

def perform
  lock = acquire_liveness_lock
  return if lock == :contended

  perform_unit
ensure
  lock.release if lock.respond_to?(:release)
end