Class: Async::Task
- Extended by:
- Forwardable
- Defined in:
- lib/async/task.rb
Overview
A task represents the state associated with the execution of an asynchronous block.
Instance Attribute Summary collapse
- #fiber ⇒ Object readonly
- #reactor ⇒ Object readonly
- #status ⇒ Object readonly
Attributes inherited from Node
#annotation, #children, #parent
Class Method Summary collapse
-
.current ⇒ Async::Task
Lookup the Task for the current fiber.
-
.current? ⇒ Async::Task?
Check if there is a task defined for the current fiber.
-
.yield {|result| ... } ⇒ Object
Yield the unerlying
result
for the task.
Instance Method Summary collapse
- #async(*args, &block) ⇒ Object
-
#finished? ⇒ Boolean
Whether we can remove this node from the reactor graph.
-
#initialize(reactor, parent = Task.current?) ⇒ Task
constructor
Create a new task.
-
#result ⇒ Object
(also: #wait)
Retrieve the current result of the task.
-
#run(*args) ⇒ Object
Resume the execution of the task.
-
#running? ⇒ Boolean
Check if the task is running.
-
#stop ⇒ void
Stop the task and all of its children.
- #to_s ⇒ Object
-
#yield ⇒ Object
Yield back to the reactor and allow other fibers to execute.
Methods inherited from Node
#annotate, #consume, #description, #print_hierarchy, #reap, #traverse
Constructor Details
#initialize(reactor, parent = Task.current?) ⇒ Task
Create a new task.
60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 |
# File 'lib/async/task.rb', line 60 def initialize(reactor, parent = Task.current?) super(parent || reactor) @reactor = reactor @status = :initialized @result = nil @condition = nil @fiber = Fiber.new do |*args| set! begin @result = yield(self, *args) @status = :complete # Async.logger.debug("Task #{self} completed normally.") rescue Stop @status = :stop # Async.logger.debug("Task #{self} stopped: #{$!}") rescue Exception => error @result = error @status = :failed # Async.logger.debug("Task #{self} failed: #{$!}") raise ensure # Async.logger.debug("Task #{self} closing: #{$!}") finish! end end end |
Instance Attribute Details
#fiber ⇒ Object (readonly)
106 107 108 |
# File 'lib/async/task.rb', line 106 def fiber @fiber end |
#reactor ⇒ Object (readonly)
97 98 99 |
# File 'lib/async/task.rb', line 97 def reactor @reactor end |
#status ⇒ Object (readonly)
110 111 112 |
# File 'lib/async/task.rb', line 110 def status @status end |
Class Method Details
.current ⇒ Async::Task
Lookup the Async::Task for the current fiber. Raise RuntimeError
if none is available.
159 160 161 |
# File 'lib/async/task.rb', line 159 def self.current Thread.current[:async_task] or raise RuntimeError, "No async task available!" end |
.current? ⇒ Async::Task?
Check if there is a task defined for the current fiber.
165 166 167 |
# File 'lib/async/task.rb', line 165 def self.current? Thread.current[:async_task] end |
.yield {|result| ... } ⇒ Object
Yield the unerlying result
for the task. If the result
is an Exception, then that result will be raised an its
exception.
43 44 45 46 47 48 49 50 51 52 53 54 55 |
# File 'lib/async/task.rb', line 43 def self.yield if block_given? result = yield else result = Fiber.yield end if result.is_a? Exception raise result else return result end end |
Instance Method Details
#async(*args, &block) ⇒ Object
122 123 124 125 126 127 128 |
# File 'lib/async/task.rb', line 122 def async(*args, &block) task = Task.new(@reactor, self, &block) task.run(*args) return task end |
#finished? ⇒ Boolean
Whether we can remove this node from the reactor graph.
177 178 179 |
# File 'lib/async/task.rb', line 177 def finished? super && @status != :running end |
#result ⇒ Object Also known as: wait
Retrieve the current result of the task. Will cause the caller to wait until result is available.
133 134 135 136 137 138 139 140 141 142 |
# File 'lib/async/task.rb', line 133 def result raise RuntimeError.new("Cannot wait on own fiber") if Fiber.current.equal?(@fiber) if running? @condition ||= Condition.new @condition.wait else Task.yield {@result} end end |
#run(*args) ⇒ Object
Resume the execution of the task.
113 114 115 116 117 118 119 120 |
# File 'lib/async/task.rb', line 113 def run(*args) if @status == :initialized @status = :running @fiber.resume(*args) else raise RuntimeError, "Task already running!" end end |
#running? ⇒ Boolean
Check if the task is running.
171 172 173 |
# File 'lib/async/task.rb', line 171 def running? @status == :running end |
#stop ⇒ void
This method returns an undefined value.
Stop the task and all of its children.
148 149 150 151 152 153 154 |
# File 'lib/async/task.rb', line 148 def stop @children.each(&:stop) if @fiber.alive? @fiber.resume(Stop.new) end end |
#to_s ⇒ Object
92 93 94 |
# File 'lib/async/task.rb', line 92 def to_s "<#{self.description} status=#{@status}>" end |
#yield ⇒ Object
Yield back to the reactor and allow other fibers to execute.
101 102 103 |
# File 'lib/async/task.rb', line 101 def yield self.sleep(0) end |