Class: CMDx::Pipeline

Inherits:
Object
  • Object
show all
Defined in:
lib/cmdx/pipeline.rb

Overview

Runs a Workflow's declared task groups. Each group selects a strategy (:sequential by default, or :parallel). A group failure halts the pipeline by echoing the failed result's signal through throw!, which bubbles up through Runtime as the workflow's own failure.

Groups may opt into batch semantics with continue_on_failure: true, in which case every task in the group runs to completion and all failures are aggregated into the workflow's errors (keyed as "TaskClass.input" for input/validation errors and "TaskClass.<status>" for bare fail! reasons) before the pipeline halts on the first failure (declaration order).

See Also:

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(workflow) ⇒ Pipeline

Returns a new instance of Pipeline.

Parameters:

  • workflow (Task) —

    workflow instance whose class includes Workflow



30
31
32
33
 
# File 'lib/cmdx/pipeline.rb', line 30

def initialize(workflow)
  @workflow = workflow
  @executed = []
end

Class Method Details

.execute(workflow) ⇒ void

This method returns an undefined value.

Parameters:

  • workflow (Task) —

    workflow instance whose class includes Workflow



23
24
25
 
# File 'lib/cmdx/pipeline.rb', line 23

def execute(workflow)
  new(workflow).execute
end

Instance Method Details

#execute ⇒ void

This method returns an undefined value.

Iterates every group in the workflow's pipeline, respecting :if/:unless and the :strategy key. Any group that produces a failed result halts execution by throwing through the workflow.

On halt, every previously executed task instance whose result is success? is sent #rollback (when defined) in reverse execution order, providing saga-style compensation. Each compensated result has its :rolled_back option flipped to true. Skipped tasks are excluded; the failing task itself is rolled back by Runtime and is not re-invoked here. Exceptions raised inside a compensator propagate — handling them is the developer's responsibility.

Raises:

  • (ArgumentError) —

    for an unknown strategy

  • (StandardError) —

    anything raised by a task's #rollback



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/cmdx/pipeline.rb', line 50

def execute
  @workflow.class.pipeline.each do |group|
    next unless Util.satisfied?(group.options[:if], group.options[:unless], @workflow)

    halt =
      case strategy = group.options[:strategy]
      when :sequential, NilClass
        run_sequential(group)
      when :parallel
        run_parallel(group)
      else
        raise ArgumentError, "invalid strategy: #{strategy.inspect}"
      end

    next unless halt

    rollback_executed!
    @workflow.send(:throw!, halt)
  end
end