Class: SimplerWorkflow::Activity
- Inherits:
-
Object
- Object
- SimplerWorkflow::Activity
show all
- Includes:
- OptionsAsMethods
- Defined in:
- lib/simpler_workflow/activity.rb
Instance Attribute Summary collapse
Class Method Summary
collapse
Instance Method Summary
collapse
#method_missing
Constructor Details
#initialize(domain, name, version, options = {}) ⇒ Activity
Returns a new instance of Activity.
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
# File 'lib/simpler_workflow/activity.rb', line 7
def initialize(domain, name, version, options = {})
default_options =
{
:default_task_list => name,
:default_task_start_to_close_timeout => 5 * 60,
:default_task_schedule_to_start_timeout => 5 * 60,
:default_task_schedule_to_close_timeout => 10 * 60,
:default_task_heartbeat_timeout => :none
}
@options = default_options.merge(options)
@domain = domain
@name = name
@version = version
@failure_policy = :fail
@task_list = name.to_s
end
|
Instance Attribute Details
#domain ⇒ Object
Returns the value of attribute domain.
5
6
7
|
# File 'lib/simpler_workflow/activity.rb', line 5
def domain
@domain
end
|
#name ⇒ Object
Returns the value of attribute name.
5
6
7
|
# File 'lib/simpler_workflow/activity.rb', line 5
def name
@name
end
|
#next_activity ⇒ Object
Returns the value of attribute next_activity.
5
6
7
|
# File 'lib/simpler_workflow/activity.rb', line 5
def next_activity
@next_activity
end
|
#options ⇒ Object
Returns the value of attribute options.
5
6
7
|
# File 'lib/simpler_workflow/activity.rb', line 5
def options
@options
end
|
#task_list ⇒ Object
Returns the value of attribute task_list.
5
6
7
|
# File 'lib/simpler_workflow/activity.rb', line 5
def task_list
@task_list
end
|
#version ⇒ Object
Returns the value of attribute version.
5
6
7
|
# File 'lib/simpler_workflow/activity.rb', line 5
def version
@version
end
|
Class Method Details
.[](*activity_tuple) ⇒ Object
163
164
165
|
# File 'lib/simpler_workflow/activity.rb', line 163
def self.[](*activity_tuple)
activities[*activity_tuple]
end
|
.[]=(*activity_tuple) ⇒ Object
167
168
169
|
# File 'lib/simpler_workflow/activity.rb', line 167
def self.[]=(*activity_tuple)
activities.[]=(*activity_tuple)
end
|
.register(domain, name, version, activity) ⇒ Object
171
172
173
|
# File 'lib/simpler_workflow/activity.rb', line 171
def self.register(domain, name, version, activity)
activities.register(domain, name, version, activity)
end
|
Instance Method Details
#count ⇒ Object
159
160
161
|
# File 'lib/simpler_workflow/activity.rb', line 159
def count
domain.activity_tasks.count(name).to_i
end
|
#failure_policy ⇒ Object
42
43
44
|
# File 'lib/simpler_workflow/activity.rb', line 42
def failure_policy
@failure_policy || :fail
end
|
#on_fail(failure_policy) ⇒ Object
38
39
40
|
# File 'lib/simpler_workflow/activity.rb', line 38
def on_fail(failure_policy)
@failure_policy = failure_policy.to_sym
end
|
#on_success(activity, version = nil) ⇒ Object
24
25
26
27
28
29
30
31
32
33
34
35
36
|
# File 'lib/simpler_workflow/activity.rb', line 24
def on_success(activity, version = nil)
case activity
when Hash
name = activity[:name].to_sym
version = activity[:version]
when String
name = activity.to_sym
when Symbol
name = activity
end
@next_activity = Activity[domain, name, version]
end
|
46
47
48
|
# File 'lib/simpler_workflow/activity.rb', line 46
def perform_activity(&block)
@perform_task = block
end
|
54
55
56
57
58
59
60
61
62
63
64
|
# File 'lib/simpler_workflow/activity.rb', line 54
def perform_task(task)
logger.info("Performing task #{name}")
@perform_task.call(task)
rescue => e
context = {}
context[:activity_type] = [name.to_s, version]
context[:input] = task.input
context[:activity_id] = task.activity_id
SimplerWorkflow.exception_reporter.report(e, context)
task.fail! :reason => e.message[0..250], :details => {:failure_policy => failure_policy}.to_json
end
|
#persist_attributes ⇒ Object
70
71
72
|
# File 'lib/simpler_workflow/activity.rb', line 70
def persist_attributes
activities.persist_attributes(self)
end
|
#poll_for_single_task ⇒ Object
154
155
156
157
|
# File 'lib/simpler_workflow/activity.rb', line 154
def poll_for_single_task
logger.info("Polling for single task for #{name}")
domain.activity_tasks.poll_for_single_task(name.to_s)
end
|
#simple_db_attributes ⇒ Object
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
# File 'lib/simpler_workflow/activity.rb', line 74
def simple_db_attributes
attributes = {
domain: domain.name,
name: name,
version: version,
failure_policy: failure_policy
}
if (next_activity)
attributes[:next_activity_name] = next_activity.name
attributes[:next_activity_version] = next_activity.version
end
attributes
end
|
#simple_db_name ⇒ Object
90
91
92
|
# File 'lib/simpler_workflow/activity.rb', line 90
def simple_db_name
"#{name}-#{version}"
end
|
#start_activity_loop ⇒ Object
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
# File 'lib/simpler_workflow/activity.rb', line 94
def start_activity_loop
SimplerWorkflow.child_processes << fork do
$0 = "Activity: #{name} #{version}"
Signal.trap('QUIT') do
if @in_task
@time_to_exit = true
else
Process.exit 0
end
end
Signal.trap('INT') do
Process.exit!(0)
end
if SimplerWorkflow.after_fork
SimplerWorkflow.after_fork.call
end
loop do
begin
logger.info("Starting activity_loop for #{name}")
domain.activity_tasks.poll(task_list) do |task|
begin
logger.info("Received task...")
@in_task = true
perform_task(task)
unless task.responded?
task.complete!
end
rescue => e
context = {}
context[:activity_type] = [name.to_s, version]
context[:input] = task.input
context[:activity_id] = task.activity_id
SimplerWorkflow.exception_reporter.report(e, context)
task.fail! :reason => e.message, :details => { :failure_policy => :fail }.to_json unless task.responded?
ensure
@in_task = false
end
end
Process.exit(0) if @time_to_exit
rescue Timeout::Error
if @time_to_exit
Process.exit(0)
else
retry
end
end
end
end
end
|
#to_activity_type ⇒ Object
66
67
68
|
# File 'lib/simpler_workflow/activity.rb', line 66
def to_activity_type
domain.activity_types[name, version]
end
|