Class: Gaskit::Flow

Inherits:
Object
  • Object
show all
Includes:
Hookable
Defined in:
lib/gaskit/flow.rb

Overview

The Gaskit::Flow class defines and executes a pipeline of operations, each of which returns a Gaskit::OperationResult. Flows can be defined via a class-based DSL or run inline using a block.

Steps share context and flow state, with support for early exits, manual control, and step-by-step walking. A flow is constructed from a sequence of registered operations, each of which may receive and return input to influence subsequent steps.

Features

  • Declarative or block-based step definitions
  • Per-step argument and context overrides
  • Shared flow-level context with metadata injection
  • Manual stepping (walk and next_step)
  • Rewind capability for retryable flows
  • Hookable lifecycle via Gaskit::Hookable

Examples:

Inline (block-based) flow

result = Gaskit::Flow.call(1) do
  step Add
  step Double
end

result.value # => 4

Class-based flow

class MyFlow < Gaskit::Flow
  step Add
  step Double
end

result = MyFlow.call(1)
result.value # => 4

Step-by-step execution (walk)

flow = MyFlow.walk(1)
while flow.has_next_step?
  flow.next_step
end
flow.result.value # => 4

Rewinding and re-running

flow.rewind
flow.next_step # starts from first step again

See Also:

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Hookable

#apply_after_hooks, #apply_around_hooks, #apply_before_hooks, #apply_hooks, included

Instance Attribute Details

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



155
156
157
 
# File 'lib/gaskit/flow.rb', line 155

def logger
  @logger
end

Class Method Details

.call(*args, context: {}, **kwargs, &block) ⇒ FlowResult

Executes the flow with soft failure handling.

Parameters:

  • Positional arguments for the first step.

  • (defaults to: {})

    Shared context across all steps.

  • Keyword arguments for the first step.

Returns:

  • The result of the flow execution.



94
95
96
 
# File 'lib/gaskit/flow.rb', line 94

def call(*args, context: {}, **kwargs, &block)
  invoke(false, context, *args, **kwargs, &block)
end

.call!(*args, context: {}, **kwargs, &block) ⇒ FlowResult

Executes the flow with hard failure handling (raises on unhandled errors).

Parameters:

  • Positional arguments for the first step.

  • (defaults to: {})

    Shared context across all steps.

  • Keyword arguments for the first step.

Returns:

  • The result of the flow execution.

Raises:

  • If an error occurs in any step.



105
106
107
 
# File 'lib/gaskit/flow.rb', line 105

def call!(*args, context: {}, **kwargs, &block)
  invoke(true, context, *args, **kwargs, &block)
end

.inherited(subclass) ⇒ void

This method returns an undefined value.

Called when a subclass is defined, initializing an empty step list.

Parameters:

  • the subclass inheriting from Flow



66
67
68
69
 
# File 'lib/gaskit/flow.rb', line 66

def inherited(subclass)
  subclass.instance_variable_set(:@steps, @steps)
  super
end

.invoke(raise_on_failure, context, *args, **kwargs, &block) ⇒ FlowResult

Internal flow execution logic.

Parameters:

  • Whether to raise on failure.

  • Execution context.

  • Initial positional arguments.

  • Initial keyword arguments.

Returns:

  • The result of the executed flow.



136
137
138
139
 
# File 'lib/gaskit/flow.rb', line 136

def invoke(raise_on_failure, context, *args, **kwargs, &block)
  flow = build(raise_on_failure, context, *args, **kwargs, &block)
  flow.send(:execute, &block)
end

.step(operation, context: {}, **kwargs) ⇒ void

This method returns an undefined value.

Registers a step in the class-level flow definition.

Parameters:

  • The operation to run.

  • (defaults to: {})

    Optional context for this step.

  • Keyword arguments for this step.



84
85
86
 
# File 'lib/gaskit/flow.rb', line 84

def step(operation, context: {}, **kwargs)
  steps << [operation, context, kwargs]
end

.steps ⇒ Array<Array>

Returns the list of declared steps for the flow class.

Returns:

  • An array of [operation, context, kwargs] triples.



74
75
76
 
# File 'lib/gaskit/flow.rb', line 74

def steps
  @steps ||= []
end

.walk(*args, context: {}, **kwargs) ⇒ Flow

Creates a flow instance for step-by-step execution.

Parameters:

  • Initial positional arguments.

  • (defaults to: {})

    Shared execution context.

  • Initial keyword arguments.

Returns:

  • A walkable flow instance.



115
116
117
 
# File 'lib/gaskit/flow.rb', line 115

def walk(*args, context: {}, **kwargs)
  build(false, context, *args, **kwargs)
end

.walk!(*args, context: {}, **kwargs) ⇒ Flow

Same as #walk but raises on any step failure.

Parameters:

  • Initial positional arguments.

  • (defaults to: {})

    Shared execution context.

  • Initial keyword arguments.

Returns:

  • A walkable flow instance with hard failure behavior.



125
126
127
 
# File 'lib/gaskit/flow.rb', line 125

def walk!(*args, context: {}, **kwargs)
  build(true, context, *args, **kwargs)
end

Instance Method Details

#next_step(*args, **kwargs) ⇒ Gaskit::OperationResult?

Executes the next step in the flow.

Parameters:

  • Optional positional arguments to override input.

  • Optional keyword arguments to override input.

Returns:

  • The result of the step, or nil if no steps remain.



186
187
188
189
190
191
192
193
 
# File 'lib/gaskit/flow.rb', line 186

def next_step(*args, **kwargs)
  return unless next_step?

  operation, context, step_kwargs = @step_sequence[@step_index]
  @step_index += 1

  process_step(operation, context, step_kwargs, args, kwargs)
end

#next_step? ⇒ Boolean

Returns true if there are more steps remaining in the sequence.

Returns:

  • Whether the flow has more steps to execute.



160
161
162
 
# File 'lib/gaskit/flow.rb', line 160

def next_step?
  @step_index < @step_sequence.size
end

#pending_step(*args, **kwargs) ⇒ Hash?

Returns the metadata for the pending step.

Returns:

  • The pending step's operation, context, and kwargs, or nil if no steps remain.



167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/gaskit/flow.rb', line 167

def pending_step(*args, **kwargs)
  return nil unless next_step?

  operation, context, step_kwargs = @step_sequence[@step_index]
  context = @context.merge(context)
  args, kwargs = resolve_step_input(
    args: args,
    kwargs: kwargs,
    step_kwargs: step_kwargs
  )

  { operation: operation, context: context, args: args, kwargs: kwargs }
end

#results ⇒ Array<Hash>

Returns the result hashes from all executed steps.

Returns:

  • Array of step metadata hashes including input/output.



218
219
220
 
# File 'lib/gaskit/flow.rb', line 218

def results
  @steps.map { |entry| entry[:result] }
end

#rewind ⇒ void

This method returns an undefined value.

Rewinds the flow to its initial state.



208
209
210
211
212
213
 
# File 'lib/gaskit/flow.rb', line 208

def rewind
  @input = @initial_input.dup
  @result = nil
  @steps.clear
  @step_index = 0
end

#step(operation, context: {}, **kwargs) ⇒ Gaskit::OperationResult

Runs a step inline from within a block-based flow definition.

Parameters:

  • The operation to execute.

  • (defaults to: {})

    Additional context for the step.

  • Keyword arguments for the step.

Returns:

  • The result of the step.



201
202
203
 
# File 'lib/gaskit/flow.rb', line 201

def step(operation, context: {}, **kwargs)
  process_step(operation, context, kwargs)
end