Class: Gush::Workflow
- Inherits:
-
Object
- Object
- Gush::Workflow
- Defined in:
- lib/gush/workflow.rb
Instance Attribute Summary collapse
-
#arguments ⇒ Object
Returns the value of attribute arguments.
-
#id ⇒ Object
Returns the value of attribute id.
-
#jobs ⇒ Object
Returns the value of attribute jobs.
-
#persisted ⇒ Object
Returns the value of attribute persisted.
-
#stopped ⇒ Object
Returns the value of attribute stopped.
Class Method Summary collapse
Instance Method Summary collapse
- #configure(*args) ⇒ Object
- #continue ⇒ Object
- #expire!(ttl = nil) ⇒ Object
- #failed? ⇒ Boolean
- #find_job(name) ⇒ Object
- #finished? ⇒ Boolean
- #finished_at ⇒ Object
- #initial_jobs ⇒ Object
-
#initialize(*args) ⇒ Workflow
constructor
A new instance of Workflow.
- #mark_as_persisted ⇒ Object
- #mark_as_started ⇒ Object
- #mark_as_stopped ⇒ Object
- #persist! ⇒ Object
- #reload ⇒ Object
- #resolve_dependencies ⇒ Object
- #run(klass, opts = {}) ⇒ Object
- #running? ⇒ Boolean
- #save ⇒ Object
- #start! ⇒ Object
- #started? ⇒ Boolean
- #started_at ⇒ Object
- #status ⇒ Object
- #stopped? ⇒ Boolean
- #to_hash ⇒ Object
- #to_json(options = {}) ⇒ Object
Constructor Details
#initialize(*args) ⇒ Workflow
Returns a new instance of Workflow.
7 8 9 10 11 12 13 14 15 16 |
# File 'lib/gush/workflow.rb', line 7 def initialize(*args) @id = id @jobs = [] @dependencies = [] @persisted = false @stopped = false @arguments = args setup end |
Instance Attribute Details
#arguments ⇒ Object
Returns the value of attribute arguments.
5 6 7 |
# File 'lib/gush/workflow.rb', line 5 def arguments @arguments end |
#id ⇒ Object
Returns the value of attribute id.
5 6 7 |
# File 'lib/gush/workflow.rb', line 5 def id @id end |
#jobs ⇒ Object
Returns the value of attribute jobs.
5 6 7 |
# File 'lib/gush/workflow.rb', line 5 def jobs @jobs end |
#persisted ⇒ Object
Returns the value of attribute persisted.
5 6 7 |
# File 'lib/gush/workflow.rb', line 5 def persisted @persisted end |
#stopped ⇒ Object
Returns the value of attribute stopped.
5 6 7 |
# File 'lib/gush/workflow.rb', line 5 def stopped @stopped end |
Class Method Details
.create(*args) ⇒ Object
22 23 24 25 26 |
# File 'lib/gush/workflow.rb', line 22 def self.create(*args) flow = new(*args) flow.save flow end |
.descendants ⇒ Object
191 192 193 |
# File 'lib/gush/workflow.rb', line 191 def self.descendants ObjectSpace.each_object(Class).select { |klass| klass < self } end |
.find(id) ⇒ Object
18 19 20 |
# File 'lib/gush/workflow.rb', line 18 def self.find(id) Gush::Client.new.find_workflow(id) end |
Instance Method Details
#configure(*args) ⇒ Object
41 42 |
# File 'lib/gush/workflow.rb', line 41 def configure(*args) end |
#continue ⇒ Object
28 29 30 31 32 33 34 35 |
# File 'lib/gush/workflow.rb', line 28 def continue client = Gush::Client.new failed_jobs = jobs.select(&:failed?) failed_jobs.each do |job| client.enqueue_job(id, job) end end |
#expire!(ttl = nil) ⇒ Object
56 57 58 |
# File 'lib/gush/workflow.rb', line 56 def expire! (ttl=nil) client.expire_workflow(self, ttl) end |
#failed? ⇒ Boolean
102 103 104 |
# File 'lib/gush/workflow.rb', line 102 def failed? jobs.any?(&:failed?) end |
#find_job(name) ⇒ Object
78 79 80 81 82 83 84 85 86 87 88 |
# File 'lib/gush/workflow.rb', line 78 def find_job(name) match_data = /(?<klass>\w*[^-])-(?<identifier>.*)/.match(name.to_s) if match_data.nil? job = jobs.find { |node| node.klass.to_s == name.to_s } else job = jobs.find { |node| node.name.to_s == name.to_s } end job end |
#finished? ⇒ Boolean
90 91 92 |
# File 'lib/gush/workflow.rb', line 90 def finished? jobs.all?(&:finished?) end |
#finished_at ⇒ Object
167 168 169 |
# File 'lib/gush/workflow.rb', line 167 def finished_at last_job ? last_job.finished_at : nil end |
#initial_jobs ⇒ Object
144 145 146 |
# File 'lib/gush/workflow.rb', line 144 def initial_jobs jobs.select(&:has_no_dependencies?) end |
#mark_as_persisted ⇒ Object
60 61 62 |
# File 'lib/gush/workflow.rb', line 60 def mark_as_persisted @persisted = true end |
#mark_as_started ⇒ Object
64 65 66 |
# File 'lib/gush/workflow.rb', line 64 def mark_as_started @stopped = false end |
#mark_as_stopped ⇒ Object
44 45 46 |
# File 'lib/gush/workflow.rb', line 44 def mark_as_stopped @stopped = true end |
#persist! ⇒ Object
52 53 54 |
# File 'lib/gush/workflow.rb', line 52 def persist! client.persist_workflow(self) end |
#reload ⇒ Object
135 136 137 138 139 140 141 142 |
# File 'lib/gush/workflow.rb', line 135 def reload flow = self.class.find(id) self.jobs = flow.jobs self.stopped = flow.stopped self end |
#resolve_dependencies ⇒ Object
68 69 70 71 72 73 74 75 76 |
# File 'lib/gush/workflow.rb', line 68 def resolve_dependencies @dependencies.each do |dependency| from = find_job(dependency[:from]) to = find_job(dependency[:to]) to.incoming << dependency[:from] from.outgoing << dependency[:to] end end |
#run(klass, opts = {}) ⇒ Object
110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
# File 'lib/gush/workflow.rb', line 110 def run(klass, opts = {}) node = klass.new({ workflow_id: id, id: client.next_free_job_id(id, klass.to_s), params: opts.fetch(:params, {}), queue: opts[:queue] }) jobs << node deps_after = [*opts[:after]] deps_after.each do |dep| @dependencies << {from: dep.to_s, to: node.name.to_s } end deps_before = [*opts[:before]] deps_before.each do |dep| @dependencies << {from: node.name.to_s, to: dep.to_s } end node.name end |
#running? ⇒ Boolean
98 99 100 |
# File 'lib/gush/workflow.rb', line 98 def running? started? && !finished? end |
#save ⇒ Object
37 38 39 |
# File 'lib/gush/workflow.rb', line 37 def save persist! end |
#start! ⇒ Object
48 49 50 |
# File 'lib/gush/workflow.rb', line 48 def start! client.start_workflow(self) end |
#started? ⇒ Boolean
94 95 96 |
# File 'lib/gush/workflow.rb', line 94 def started? !!started_at end |
#started_at ⇒ Object
163 164 165 |
# File 'lib/gush/workflow.rb', line 163 def started_at first_job ? first_job.started_at : nil end |
#status ⇒ Object
148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
# File 'lib/gush/workflow.rb', line 148 def status case when failed? :failed when running? :running when finished? :finished when stopped? :stopped else :running end end |
#stopped? ⇒ Boolean
106 107 108 |
# File 'lib/gush/workflow.rb', line 106 def stopped? stopped end |
#to_hash ⇒ Object
171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 |
# File 'lib/gush/workflow.rb', line 171 def to_hash name = self.class.to_s { name: name, id: id, arguments: @arguments, total: jobs.count, finished: jobs.count(&:finished?), klass: name, status: status, stopped: stopped, started_at: started_at, finished_at: finished_at } end |