Class: Floe::Workflow::State

Inherits:
Object
  • Object
show all
Includes:
ValidationMixin
Defined in:
lib/floe/workflow/state.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from ValidationMixin

included, #invalid_field_error!, #missing_field_error!, #parser_error!, #runtime_field_error!, #wrap_parser_error

Constructor Details

#initialize(_workflow, name, payload) ⇒ State

Returns a new instance of State.



26
27
28
29
30
31
 
# File 'lib/floe/workflow/state.rb', line 26

def initialize(_workflow, name, payload)
  @name     = name
  @payload  = payload
  @type     = payload["Type"]
  @comment  = payload["Comment"]
end

Instance Attribute Details

#comment ⇒ Object (readonly)

Returns the value of attribute comment.



24
25
26
 
# File 'lib/floe/workflow/state.rb', line 24

def comment
  @comment
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



24
25
26
 
# File 'lib/floe/workflow/state.rb', line 24

def name
  @name
end

#payload ⇒ Object (readonly)

Returns the value of attribute payload.



24
25
26
 
# File 'lib/floe/workflow/state.rb', line 24

def payload
  @payload
end

#type ⇒ Object (readonly)

Returns the value of attribute type.



24
25
26
 
# File 'lib/floe/workflow/state.rb', line 24

def type
  @type
end

Class Method Details

.build!(workflow, name, payload) ⇒ Object 



9
10
11
12
13
14
15
16
17
18
19
20
21
 
# File 'lib/floe/workflow/state.rb', line 9

def build!(workflow, name, payload)
  state_type = payload["Type"]
  missing_field_error!(name, "Type") if payload["Type"].nil?
  invalid_field_error!(name[0..-2], "Name", name.last, "must be less than or equal to 80 characters") if name.last.length > 80

  begin
    klass = Floe::Workflow::States.const_get(state_type)
  rescue NameError
    invalid_field_error!(name, "Type", state_type, "is not valid")
  end

  klass.new(workflow, name, payload)
end

Instance Method Details

#finish(context) ⇒ Object 



64
65
66
 
# File 'lib/floe/workflow/state.rb', line 64

def finish(context)
  mark_finished(context)
end

#finished?(context) ⇒ Boolean

Returns:

  • (Boolean) 


68
69
70
 
# File 'lib/floe/workflow/state.rb', line 68

def finished?(context)
  context.state_finished?
end

#long_name ⇒ Object 



122
123
124
 
# File 'lib/floe/workflow/state.rb', line 122

def long_name
  "#{type}:#{short_name}"
end

#mark_error(context, exception) ⇒ Object 



91
92
93
94
95
96
97
 
# File 'lib/floe/workflow/state.rb', line 91

def mark_error(context, exception)
  # InputPath or OutputPath were bad.
  context.next_state = nil
  context.output     = exception.to_output
  # Since finish threw an exception, super was never called. Calling that now.
  mark_finished(context)
end

#mark_finished(context) ⇒ Object 



78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/floe/workflow/state.rb', line 78

def mark_finished(context)
  finished_time = Time.now.utc
  entered_time  = Time.parse(context.state["EnteredTime"])

  context.state["FinishedTime"] ||= finished_time.iso8601
  context.state["Duration"]       = finished_time - entered_time

  level = context.failed? ? :error : :info
  context.logger.public_send(level, "Running state: [#{long_name}] with input [#{context.json_input}]...Complete #{context.next_state ? "- next state [#{context.next_state}]" : "workflow -"} output: [#{context.json_output}]")

  0
end

#mark_started(context) ⇒ Object 



72
73
74
75
76
 
# File 'lib/floe/workflow/state.rb', line 72

def mark_started(context)
  context.state["EnteredTime"] = Time.now.utc.iso8601

  context.logger.info("Running state: [#{long_name}] with input [#{context.json_input}]...")
end

#ready?(context) ⇒ Boolean

Returns:

  • (Boolean) 


99
100
101
102
103
104
 
# File 'lib/floe/workflow/state.rb', line 99

def ready?(context)
  return false if !started?(context)
  return true  if !running?(context)
  return true  if wait_until(context) && !waiting?(context)
  false
end

#run_nonblock!(context) ⇒ Object

Returns for incomplete Errno::EAGAIN, for completed 0.

Returns:

  • for incomplete Errno::EAGAIN, for completed 0



45
46
47
48
49
50
51
52
53
54
 
# File 'lib/floe/workflow/state.rb', line 45

def run_nonblock!(context)
  # Only start the state if it isn't already started and it isn't waiting
  # from a prior Retry.
  start(context)       unless started?(context) || waiting?(context)
  return Errno::EAGAIN unless ready?(context)

  finish(context)
rescue Floe::ExecutionError => e
  mark_error(context, e)
end

#running?(context) ⇒ Boolean

Returns:

  • (Boolean)

Raises:

  • (NotImplementedError) 


106
107
108
 
# File 'lib/floe/workflow/state.rb', line 106

def running?(context)
  raise NotImplementedError, "Must be implemented in a subclass"
end

#short_name ⇒ Object 



118
119
120
 
# File 'lib/floe/workflow/state.rb', line 118

def short_name
  name.last
end

#start(context) ⇒ Object 



56
57
58
 
# File 'lib/floe/workflow/state.rb', line 56

def start(context)
  mark_started(context)
end

#started?(context) ⇒ Boolean

Returns:

  • (Boolean) 


60
61
62
 
# File 'lib/floe/workflow/state.rb', line 60

def started?(context)
  context.state_started?
end

#wait(context, timeout: nil) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
 
# File 'lib/floe/workflow/state.rb', line 33

def wait(context, timeout: nil)
  start = Time.now.utc

  loop do
    return 0             if ready?(context)
    return Errno::EAGAIN if timeout && (timeout.zero? || Time.now.utc - start > timeout)

    sleep(1)
  end
end

#wait_until(context) ⇒ Object 



114
115
116
 
# File 'lib/floe/workflow/state.rb', line 114

def wait_until(context)
  context.state["WaitUntil"] && Time.parse(context.state["WaitUntil"])
end

#waiting?(context) ⇒ Boolean

Returns:

  • (Boolean) 


110
111
112
 
# File 'lib/floe/workflow/state.rb', line 110

def waiting?(context)
  context.state["WaitUntil"] && Time.now.utc <= Time.parse(context.state["WaitUntil"])
end