Class: RubyReactor::Adapters::Sidekiq::Router

Inherits:
Object
  • Object
show all
Defined in:
lib/ruby_reactor/adapters/sidekiq/router.rb

Class Method Summary collapse

Class Method Details

.perform_async(context_id, reactor_class_name = nil, intermediate_results: {}) ⇒ Object

Identity-only payload: the worker rehydrates the live context from storage by (context_id, reactor_class_name). The caller already holds context_id, so there is no blob to deserialize here.



10
11
12
13
14
 
# File 'lib/ruby_reactor/adapters/sidekiq/router.rb', line 10

def self.perform_async(context_id, reactor_class_name = nil, intermediate_results: {})
  job_id = RubyReactor::Adapters::Sidekiq::Worker.perform_async(context_id, reactor_class_name)
  RubyReactor::DispatchResult.new(job_id: job_id, intermediate_results: intermediate_results,
                               execution_id: context_id)
end

.perform_in(delay, context_id, reactor_class_name = nil, intermediate_results: {}) ⇒ Object 



16
17
18
19
20
 
# File 'lib/ruby_reactor/adapters/sidekiq/router.rb', line 16

def self.perform_in(delay, context_id, reactor_class_name = nil, intermediate_results: {})
  job_id = RubyReactor::Adapters::Sidekiq::Worker.perform_in(delay, context_id, reactor_class_name)
  RubyReactor::DispatchResult.new(job_id: job_id, intermediate_results: intermediate_results,
                               execution_id: context_id)
end

.perform_map_collection_async(parent_context_id:, map_id:, parent_reactor_class_name:, step_name:, strict_ordering:, timeout:) ⇒ Object

rubocop:disable Metrics/ParameterLists



88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/ruby_reactor/adapters/sidekiq/router.rb', line 88

def self.perform_map_collection_async(parent_context_id:, map_id:, parent_reactor_class_name:, step_name:,
                                      strict_ordering:, timeout:)
  job_id = RubyReactor::Adapters::Sidekiq::MapCollectorWorker.perform_async(
    {
      "parent_context_id" => parent_context_id,
      "map_id" => map_id,
      "parent_reactor_class_name" => parent_reactor_class_name,
      "step_name" => step_name,
      "strict_ordering" => strict_ordering,
      "timeout" => timeout
    }
  )
  RubyReactor::DispatchResult.new(job_id: job_id)
end

.perform_map_element_async(map_id:, element_id:, index:, serialized_inputs:, reactor_class_info:, strict_ordering:, parent_context_id:, parent_reactor_class_name:, step_name:, batch_size: nil, serialized_context: nil, fail_fast: nil) ⇒ Object

rubocop:disable Metrics/ParameterLists



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/ruby_reactor/adapters/sidekiq/router.rb', line 38

def self.perform_map_element_async(map_id:, element_id:, index:, serialized_inputs:, reactor_class_info:,
                                   strict_ordering:, parent_context_id:, parent_reactor_class_name:,
                                   step_name:, batch_size: nil, serialized_context: nil, fail_fast: nil)
  job_id = RubyReactor::Adapters::Sidekiq::MapElementWorker.perform_async(
    {
      "map_id" => map_id,
      "element_id" => element_id,
      "index" => index,
      "serialized_inputs" => serialized_inputs,
      "reactor_class_info" => reactor_class_info,
      "strict_ordering" => strict_ordering,
      "parent_context_id" => parent_context_id,
      "parent_reactor_class_name" => parent_reactor_class_name,
      "step_name" => step_name,
      "batch_size" => batch_size,
      "serialized_context" => serialized_context,
      "fail_fast" => fail_fast
    }
  )
  RubyReactor::DispatchResult.new(job_id: job_id)
end

.perform_map_element_in(delay, map_id:, element_id:, index:, serialized_inputs:, reactor_class_info:, strict_ordering:, parent_context_id:, parent_reactor_class_name:, step_name:, batch_size: nil, serialized_context: nil, fail_fast: nil) ⇒ Object 



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/ruby_reactor/adapters/sidekiq/router.rb', line 60

def self.perform_map_element_in(delay, map_id:, element_id:, index:, serialized_inputs:,
                                reactor_class_info:, strict_ordering:, parent_context_id:,
                                parent_reactor_class_name:, step_name:, batch_size: nil,
                                serialized_context: nil, fail_fast: nil)
  job_id = RubyReactor::Adapters::Sidekiq::MapElementWorker.perform_in(
    delay,
    {
      "map_id" => map_id,
      "element_id" => element_id,
      "index" => index,
      "serialized_inputs" => serialized_inputs,
      "reactor_class_info" => reactor_class_info,
      "strict_ordering" => strict_ordering,
      "parent_context_id" => parent_context_id,
      "parent_reactor_class_name" => parent_reactor_class_name,
      "step_name" => step_name,
      "batch_size" => batch_size,
      "serialized_context" => serialized_context,
      "fail_fast" => fail_fast
    }
  )
  # Return an DispatchResult so RetryManager#handle_async_retry recognises the
  # element was successfully requeued and yields a RetryQueuedResult.
  RubyReactor::DispatchResult.new(job_id: job_id)
end

.perform_step_async(root_context_id:, reactor_class_name:, step_context_id:, step_name:) ⇒ Object

One async_step's work, dispatched as its own independent unit. The payload is identity-only for the same reason every other dispatch here is: the worker rehydrates the parent context from storage.



25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/ruby_reactor/adapters/sidekiq/router.rb', line 25

def self.perform_step_async(root_context_id:, reactor_class_name:, step_context_id:, step_name:)
  job_id = RubyReactor::Adapters::Sidekiq::StepWorker.perform_async(
    {
      "root_context_id" => root_context_id,
      "reactor_class_name" => reactor_class_name,
      "step_context_id" => step_context_id,
      "step_name" => step_name.to_s
    }
  )
  RubyReactor::DispatchResult.new(job_id: job_id)
end