Class: JobWorkflow::SubTaskJob

Inherits:
ActiveJob::Base
  • Object
show all
Includes:
ActiveJob::Continuable
Defined in:
lib/job_workflow/sub_task_job.rb,
sig/generated/job_workflow/sub_task_job.rbs

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#_context ⇒ Context?

Signature:

  • Context?

Returns:



48
49
50
 
# File 'lib/job_workflow/sub_task_job.rb', line 48

def _context
  @_context
end

#serialized_job_workflow_context ⇒ Hash[String, untyped]?

Signature:

  • Hash[String, untyped]?

Returns:

  • (Hash[String, untyped], nil) 


63
64
65
 
# File 'lib/job_workflow/sub_task_job.rb', line 63

def serialized_job_workflow_context
  @serialized_job_workflow_context
end

Class Method Details

.from_parent_context(context:) ⇒ SubTaskJob

: (context: Context) -> SubTaskJob

Parameters:

Returns:



9
10
11
12
13
14
15
16
17
18
19
20
 
# File 'lib/job_workflow/sub_task_job.rb', line 9

def from_parent_context(context:)
  validate_sub_task_context!(context)

  new_context = context.dup
  job = new(context.arguments.to_h)
  new_context._job = job
  job._context = new_context
  task = new_context._task_context.task || (raise "task is not set")
  return job if task.enqueue.queue.nil?

  job.set(queue: task.enqueue.queue)
end

Instance Method Details

#build_context(payload) ⇒ Context

: (Hash[Symbol, untyped]) -> Context

Parameters:

  • (Hash[Symbol, untyped])

Returns:



66
67
68
69
70
71
72
73
74
75
 
# File 'lib/job_workflow/sub_task_job.rb', line 66

def build_context(payload)
  context_data = extract_context_data(payload)
  parent_job_id = context_data.fetch("task_context").fetch("parent_job_id")
  parent_job_data = QueueAdapter.current.find_job(parent_job_id)
  raise WorkflowStatus::NotFoundError, "Workflow with job_id '#{parent_job_id}' not found" if parent_job_data.nil?

  workflow = resolve_workflow(parent_job_data.fetch("class_name"))
  Context.deserialize(context_data.merge("job" => self, "workflow" => workflow))
         ._update_arguments(payload.except(:job_workflow_context))
end

#deserialize(job_data) ⇒ void

This method returns an undefined value.

: (Hash[String, untyped]) -> void

Parameters:

  • (Hash[String, untyped]) 


56
57
58
59
 
# File 'lib/job_workflow/sub_task_job.rb', line 56

def deserialize(job_data)
  super
  self.serialized_job_workflow_context = job_data["job_workflow_context"]
end

#extract_context_data(payload) ⇒ Hash[String, untyped]

: (Hash[Symbol, untyped]) -> Hash[String, untyped]

Parameters:

  • (Hash[Symbol, untyped])

Returns:

  • (Hash[String, untyped]) 


87
88
89
90
91
92
 
# File 'lib/job_workflow/sub_task_job.rb', line 87

def extract_context_data(payload)
  context_data = serialized_job_workflow_context || payload[:job_workflow_context]
  raise "job_workflow_context is not set." if context_data.nil?

  context_data.deep_stringify_keys
end

#output ⇒ Output

: () -> Output

Returns:



41
42
43
44
45
46
 
# File 'lib/job_workflow/sub_task_job.rb', line 41

def output
  context = _context
  raise "context is not set." if context.nil?

  context.output
end

#perform(arguments) ⇒ void

This method returns an undefined value.

: (Hash[untyped, untyped]) -> void

Parameters:

  • (Hash[untyped, untyped]) 


33
34
35
36
37
38
 
# File 'lib/job_workflow/sub_task_job.rb', line 33

def perform(arguments)
  payload = arguments.symbolize_keys
  self._context = build_context(payload)
  context = _context || (raise "context is not set.")
  Runner.new(context:).run
end

#resolve_workflow(job_class_name) ⇒ Workflow

: (String job_class_name) -> Workflow

Parameters:

  • job_class_name (String)

Returns:



78
79
80
81
82
83
84
 
# File 'lib/job_workflow/sub_task_job.rb', line 78

def resolve_workflow(job_class_name)
  job_class = JobWorkflow::DSL._included_classes.to_a.reverse.find { |klass| klass.name == job_class_name }
  job_class ||= job_class_name.safe_constantize
  raise NameError, "uninitialized constant #{job_class_name}" if job_class.nil?

  job_class._workflow
end

#serialize ⇒ Hash[String, untyped]

: () -> Hash[String, untyped]

Returns:

  • (Hash[String, untyped]) 


51
52
53
 
# File 'lib/job_workflow/sub_task_job.rb', line 51

def serialize
  super.merge({ "job_workflow_context" => _context&.serialize || serialized_job_workflow_context }.compact)
end