Module: Plumbing::Operation::DSL

Included in:
Plumbing::Operation
Defined in:
lib/plumbing/operation/dsl.rb

Overview

Class-level authoring DSL, extended onto Task. This file carries the attribute mechanism; state builders are added in a later task.

Instance Method Summary collapse

Instance Method Details

#action(name, &body) ⇒ Object 



31
32
33
34
 
# File 'lib/plumbing/operation/dsl.rb', line 31

def action(name, &body)
  states[name.to_sym] = State.new(name: name.to_sym, kind: :action, action: body)
  ActionBuilder.new(self, name.to_sym)
end

#attribute(name, type, **opts) ⇒ Object

Declare a typed attribute. Adds a prop to the schema and defines delegating reader/writer methods on instances.



18
19
20
21
22
23
 
# File 'lib/plumbing/operation/dsl.rb', line 18

def attribute(name, type, **opts)
  attributes_schema.prop(name, type, **opts)
  define_method(name) { @attributes.public_send(name) }
  define_method(:"#{name}=") { |value| @attributes.public_send(:"#{name}=", value) }
  name.to_sym
end

#attributes_schema ⇒ Object

The per-class Literal::Struct that holds attribute values. Mutable, so actions can assign (self.x = ...).



12
13
14
 
# File 'lib/plumbing/operation/dsl.rb', line 12

def attributes_schema
  @attributes_schema ||= Class.new(Literal::Struct)
end

#call(pipeline: nil, **attrs) ⇒ Object 



64
65
66
67
 
# File 'lib/plumbing/operation/dsl.rb', line 64

def call(pipeline: nil, **attrs)
  ensure_worker_supports_waits!
  new(pipeline: pipeline).tap { |op| op.__send__(:start, attrs) }
end

#decision(name, &block) ⇒ Object 



36
37
38
39
40
41
 
# File 'lib/plumbing/operation/dsl.rb', line 36

def decision(name, &block)
  builder = DecisionBuilder.new
  builder.instance_eval(&block)
  states[name.to_sym] = State.new(name: name.to_sym, kind: :decision, transitions: builder.transitions.freeze)
  name.to_sym
end

#default_delay ⇒ Object 



50
 
# File 'lib/plumbing/operation/dsl.rb', line 50

def default_delay = @default_delay ||= 10.0

#default_timeout ⇒ Object 



54
 
# File 'lib/plumbing/operation/dsl.rb', line 54

def default_timeout = @default_timeout ||= 86_400.0

#delay(seconds) ⇒ Object 



48
 
# File 'lib/plumbing/operation/dsl.rb', line 48

def delay(seconds) = @default_delay = seconds.to_f

#ensure_worker_supports_waits! ⇒ Object

Raises:



76
77
78
79
80
 
# File 'lib/plumbing/operation/dsl.rb', line 76

def ensure_worker_supports_waits!
  return unless has_waits?
  return unless Plumbing::Actor.selected_worker_type == :inline
  raise Plumbing::Actor::NotSupported, "#{name || "operation"} has wait states; select a non-inline worker with Plumbing::Actor.uses :async (or :threaded)"
end

#has_waits? ⇒ Boolean

Returns:

  • (Boolean) 


74
 
# File 'lib/plumbing/operation/dsl.rb', line 74

def has_waits? = states.each_value.any? { |state| state.kind == :wait }

#interaction(name, &body) ⇒ Object 



89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/plumbing/operation/dsl.rb', line 89

def interaction(name, &body)
  name = name.to_sym
  define_method(name) do |*args, **kwargs|
    worker.post(name, args: args, kwargs: kwargs)
  end
  define_method(:"_#{name}") do |args:, kwargs:|
    expected = self.class.interaction_states[name]
    raise Plumbing::Operation::InvalidState, "##{name} cannot run in state #{@current_state.inspect}" unless @current_state == expected
    instance_exec(*args, **kwargs, &body)
    run_loop
  end
  InteractionBuilder.new(self, name)
end

#interaction_states ⇒ Object 



87
 
# File 'lib/plumbing/operation/dsl.rb', line 87

def interaction_states = @interaction_states ||= {}

#restore(state:, pipeline: nil, wait_elapsed: 0.0, **attrs) ⇒ Object 



82
83
84
85
 
# File 'lib/plumbing/operation/dsl.rb', line 82

def restore(state:, pipeline: nil, wait_elapsed: 0.0, **attrs)
  ensure_worker_supports_waits!
  new(pipeline: pipeline).tap { |op| op.__send__(:resume, state.to_sym, attrs, wait_elapsed.to_f) }
end

#result(name) ⇒ Object 



43
44
45
46
 
# File 'lib/plumbing/operation/dsl.rb', line 43

def result(name)
  states[name.to_sym] = State.new(name: name.to_sym, kind: :result)
  name.to_sym
end

#start_state ⇒ Object 



29
 
# File 'lib/plumbing/operation/dsl.rb', line 29

def start_state = @start_state

#starts_with(name) ⇒ Object 



27
 
# File 'lib/plumbing/operation/dsl.rb', line 27

def starts_with(name) = @start_state = name.to_sym

#states ⇒ Object 



25
 
# File 'lib/plumbing/operation/dsl.rb', line 25

def states = @states ||= {}

#test(state, pipeline: nil, **attrs) ⇒ Object 



69
70
71
72
 
# File 'lib/plumbing/operation/dsl.rb', line 69

def test(state, pipeline: nil, **attrs)
  ensure_worker_supports_waits!
  new(pipeline: pipeline).tap { |op| op.__send__(:start_at, state.to_sym, attrs) }
end

#timeout(seconds) ⇒ Object 



52
 
# File 'lib/plumbing/operation/dsl.rb', line 52

def timeout(seconds) = @default_timeout = seconds.to_f

#wait_until(name, delay: nil, timeout: nil, &block) ⇒ Object 



56
57
58
59
60
61
62
 
# File 'lib/plumbing/operation/dsl.rb', line 56

def wait_until(name, delay: nil, timeout: nil, &block)
  builder = DecisionBuilder.new
  builder.instance_eval(&block)
  options = WaitOptions.new(delay: (delay || default_delay).to_f, timeout: (timeout || default_timeout).to_f)
  states[name.to_sym] = State.new(name: name.to_sym, kind: :wait, transitions: builder.transitions.freeze, wait_options: options)
  name.to_sym
end