Class: Fractor::Workflow::ChainBuilder

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

Overview

Fluent API for building linear chain workflows. Simplifies creation of sequential processing pipelines.

Examples:

Using chain builder

workflow = Fractor::Workflow.chain("text-pipeline")
  .step("uppercase", UppercaseWorker)
  .step("reverse", ReverseWorker)
  .step("finalize", FinalizeWorker)
  .build

instance = workflow.new
result = instance.execute(input: data)

Using define class method

workflow = Fractor::Workflow::ChainBuilder.define("text-pipeline") do |chain|
  chain.step("uppercase", UppercaseWorker)
  chain.step("reverse", ReverseWorker)
  chain.step("finalize", FinalizeWorker)
end

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name) ⇒ ChainBuilder

Returns a new instance of ChainBuilder.



45
46
47
48
49
50
 
# File 'lib/fractor/workflow/chain_builder.rb', line 45

def initialize(name)
  @name = name
  @steps = []
  @input_type_class = nil
  @output_type_class = nil
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



25
26
27
 
# File 'lib/fractor/workflow/chain_builder.rb', line 25

def name
  @name
end

#steps ⇒ Object (readonly)

Returns the value of attribute steps.



25
26
27
 
# File 'lib/fractor/workflow/chain_builder.rb', line 25

def steps
  @steps
end

Class Method Details

.define(name) {|ChainBuilder| ... } ⇒ Class

Define a chain workflow using a block. This is a convenience method that creates and builds a ChainBuilder.

Examples:

workflow = Fractor::Workflow::ChainBuilder.define("my-chain") do |chain|
  chain.step("process", MyWorker)
  chain.step("finalize", FinalizeWorker)
end

Parameters:

  • name (String) —

    The workflow name

Yields:

Returns:

  • (Class) —

    A new Workflow subclass



39
40
41
42
43
 
# File 'lib/fractor/workflow/chain_builder.rb', line 39

def self.define(name, &block)
  builder = new(name)
  builder.instance_eval(&block) if block
  builder.build
end

Instance Method Details

#build ⇒ Class

Build the workflow class

Returns:

  • (Class) —

    A new Workflow subclass



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/fractor/workflow/chain_builder.rb', line 91

def build
  chain_name = @name
  chain_steps = @steps.dup
  chain_input_type = @input_type_class
  chain_output_type = @output_type_class

  Class.new(Workflow) do
    workflow chain_name do
      input_type chain_input_type if chain_input_type
      output_type chain_output_type if chain_output_type

      # Build jobs sequentially
      chain_steps.each_with_index do |step_config, index|
        step_name = step_config[:name]
        step_worker = step_config[:worker]
        step_workers = step_config[:workers]
        step_condition = step_config[:condition]

        # Determine dependencies
        needs_job = index.positive? ? chain_steps[index - 1][:name] : nil

        job step_name, step_worker,
            needs: needs_job,
            workers: step_workers,
            condition: step_condition
      end
    end
  end
end

#build! ⇒ Class

Validate and build in one step

Returns:

  • (Class) —

    A new Workflow subclass

Raises:

  • (ArgumentError) —

    if the chain is invalid



125
126
127
128
 
# File 'lib/fractor/workflow/chain_builder.rb', line 125

def build!
  validate!
  build
end

#input_type(klass) ⇒ ChainBuilder

Set the input type for the workflow

Parameters:

  • klass (Class) —

    The input type class

Returns:



56
57
58
59
 
# File 'lib/fractor/workflow/chain_builder.rb', line 56

def input_type(klass)
  @input_type_class = klass
  self
end

#output_type(klass) ⇒ ChainBuilder

Set the output type for the workflow

Parameters:

  • klass (Class) —

    The output type class

Returns:



65
66
67
68
 
# File 'lib/fractor/workflow/chain_builder.rb', line 65

def output_type(klass)
  @output_type_class = klass
  self
end

#step(name, worker, workers: nil, condition: nil) ⇒ ChainBuilder

Add a step to the chain

Parameters:

  • name (String, Symbol) —

    The step name

  • worker (Class) —

    The worker class for this step

  • workers (Integer) (defaults to: nil) —

    Optional number of parallel workers

  • condition (Proc) (defaults to: nil) —

    Optional conditional execution

Returns:



77
78
79
80
81
82
83
84
85
86
 
# File 'lib/fractor/workflow/chain_builder.rb', line 77

def step(name, worker, workers: nil, condition: nil)
  step_config = {
    name: name.to_s,
    worker: worker,
    workers: workers,
    condition: condition,
  }
  @steps << step_config
  self
end

#validate! ⇒ Object

Validate the chain configuration

Raises:

  • (ArgumentError) —

    if validation fails



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/fractor/workflow/chain_builder.rb', line 133

def validate!
  if @name.nil? || @name.empty?
    raise ArgumentError,
          "Chain must have a name"
  end

  if @steps.empty?
    raise ArgumentError,
          "Chain must have at least one step"
  end

  # Check for duplicate step names
  step_names = @steps.map { |s| s[:name] }
  duplicates = step_names.select { |n| step_names.count(n) > 1 }.uniq
  if duplicates.any?
    raise ArgumentError,
          "Duplicate step names: #{duplicates.join(', ')}"
  end

  # Validate workers
  @steps.each do |step_config|
    unless step_config[:worker]
      raise ArgumentError,
            "Step '#{step_config[:name]}' must specify a worker class"
    end

    unless step_config[:worker] < Fractor::Worker
      raise ArgumentError,
            "Step '#{step_config[:name]}' worker must inherit from Fractor::Worker"
    end
  end

  true
end