Class: Fractor::Workflow::WorkflowExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/workflow/workflow_executor.rb

Overview

Orchestrates workflow execution by managing job execution order and data flow. Refactored to use focused helper classes for each responsibility.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(workflow, input, correlation_id: nil, logger: nil, trace: false, dead_letter_queue: nil) ⇒ WorkflowExecutor

Initialize the workflow executor.

Parameters:

  • workflow (Workflow) —

    The workflow instance to execute

  • input (Object) —

    The input data for the workflow

  • correlation_id (String, nil) (defaults to: nil) —

    Optional correlation ID for tracking

  • logger (Logger, nil) (defaults to: nil) —

    Optional logger instance

  • trace (Boolean) (defaults to: false) —

    Whether to enable execution tracing

  • dead_letter_queue (DeadLetterQueue, nil) (defaults to: nil) —

    Optional dead letter queue



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 31

def initialize(workflow, input, correlation_id: nil, logger: nil,
               trace: false, dead_letter_queue: nil)
  @workflow = workflow
  @correlation_id = correlation_id
  @context = WorkflowContext.new(
    input,
    correlation_id: correlation_id,
    logger: logger,
  )
  @completed_jobs = Set.new
  @failed_jobs = Set.new
  @hooks = ExecutionHooks.new
  @trace = trace ? create_trace : nil
  @circuit_breakers = CircuitBreakerRegistry.new
  @dead_letter_queue = dead_letter_queue
  @pre_execution_context = PreExecutionContext.new(workflow, input)

  # Initialize helper classes
  @logger = WorkflowExecutionLogger.new(logger)
  @job_executor = JobExecutor.new(@context, @logger,
                                  workflow: workflow,
                                  completed_jobs: @completed_jobs,
                                  failed_jobs: @failed_jobs,
                                  dead_letter_queue: @dead_letter_queue,
                                  circuit_breakers: @circuit_breakers)
  @fallback_handler = FallbackJobHandler.new(@workflow, @context, @hooks,
                                             @logger)
end

Instance Attribute Details

#completed_jobs ⇒ Object (readonly)

Returns the value of attribute completed_jobs.



20
21
22
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 20

def completed_jobs
  @completed_jobs
end

#context ⇒ Object (readonly)

Returns the value of attribute context.



20
21
22
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 20

def context
  @context
end

#failed_jobs ⇒ Object (readonly)

Returns the value of attribute failed_jobs.



20
21
22
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 20

def failed_jobs
  @failed_jobs
end

#hooks ⇒ Object (readonly)

Returns the value of attribute hooks.



20
21
22
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 20

def hooks
  @hooks
end

#job_executor ⇒ Object (readonly)

Returns the value of attribute job_executor.



20
21
22
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 20

def job_executor
  @job_executor
end

#pre_execution_context ⇒ Object (readonly)

Returns the value of attribute pre_execution_context.



20
21
22
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 20

def pre_execution_context
  @pre_execution_context
end

#trace ⇒ Object (readonly)

Returns the value of attribute trace.



20
21
22
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 20

def trace
  @trace
end

#workflow ⇒ Object (readonly)

Returns the value of attribute workflow.



20
21
22
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 20

def workflow
  @workflow
end

Instance Method Details

#execute ⇒ WorkflowResult

Execute the workflow and return the result.

Returns:



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 63

def execute
  # Run pre-execution validation
  @pre_execution_context.validate!

  @logger.workflow_start(@workflow.class.workflow_name,
                         @context.correlation_id)
  @hooks.trigger(:workflow_start, @workflow)
  @trace&.start_job(
    job_name: "workflow",
    worker_class: @workflow.class.name,
  )

  resolver = DependencyResolver.new(@workflow.class.jobs)
  execution_order = resolver.execution_order
  start_time = Time.now

  execution_order.each do |job_group|
    execute_job_group(job_group)
    break if workflow_terminated?
  end

  end_time = Time.now
  @trace&.complete!

  @logger.workflow_complete(@workflow.class.workflow_name,
                            end_time - start_time,
                            jobs_completed: @completed_jobs.size,
                            jobs_failed: @failed_jobs.size)

  result_builder = ResultBuilder.new(@workflow, @context, @completed_jobs,
                                     @failed_jobs, trace: @trace)
  result = result_builder.build(start_time, end_time)
  @hooks.trigger(:workflow_complete, result)
  result
end

#on(event) ⇒ Object

Register a hook for workflow/job lifecycle events.

Parameters:

  • event (Symbol) —

    The event to hook into

  • block (Proc) —

    The callback to execute



103
104
105
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 103

def on(event, &)
  @hooks.register(event, &)
end

#validate_before_execution(name = nil) {|context| ... } ⇒ Object

Register a custom pre-execution validation hook. The hook receives the PreExecutionContext and can add errors/warnings.

Examples:

Add custom validation

executor.validate_before_execution(:check_api_key) do |ctx|
  unless ctx.input.api_key
    ctx.add_error("API key is required")
  end
end

Parameters:

  • name (String, Symbol) (defaults to: nil) —

    Optional name for the validation

Yields:

  • (context) —

    Block that receives the pre-execution context



119
120
121
 
# File 'lib/fractor/workflow/workflow_executor.rb', line 119

def validate_before_execution(name = nil, &)
  @pre_execution_context.add_validation_hook(name, &)
end