Class: Dispatch::Job

Inherits:
Object
  • Object
show all
Defined in:
lib/dispatch/job.rb

Overview

Track completion and return values of asynchronous requests Duck-type join and value from Thread

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(queue = Dispatch::Queue.concurrent, &block) ⇒ Job

Returns a new instance of Job.



9
10
11
12
13
14
 
# File 'lib/dispatch/job.rb', line 9

def initialize(queue = Dispatch::Queue.concurrent, &block)
  @queue = queue
  @group = Group.new
  @values = synchronize([])
  add(&block) if not block.nil?
end

Instance Attribute Details

#group ⇒ Object (readonly)

Create a Job that asynchronously dispatches the block



7
8
9
 
# File 'lib/dispatch/job.rb', line 7

def group
  @group
end

#values ⇒ Object (readonly)

Create a Job that asynchronously dispatches the block



7
8
9
 
# File 'lib/dispatch/job.rb', line 7

def values
  @values
end

Instance Method Details

#add(&block) ⇒ Object

Submit block as part of the same dispatch group



21
22
23
 
# File 'lib/dispatch/job.rb', line 21

def add(&block)
  @queue.async(@group) { @values << block.call }      
end

#join(queue = Dispatch::Queue.concurrent, &block) ⇒ Object

Wait until execution has completed. If a block is passed, invoke that asynchronously on the specified queue (or else the default queue).



28
29
30
31
 
# File 'lib/dispatch/job.rb', line 28

def join(queue = Dispatch::Queue.concurrent, &block)
  return group.wait if block.nil?
  group.notify(queue) { block.call } 
end

#synchronize(obj) ⇒ Object Also known as: sync 



16
17
18
 
# File 'lib/dispatch/job.rb', line 16

def synchronize(obj)
  Dispatch::Proxy.new(obj, @group)
end

#value(queue = Dispatch::Queue.concurrent, &block) ⇒ Object

Wait then return the next value; note: only ordered if a serial queue If a block is passed, invoke that asynchronously with the value on the specified queue (or else the default queue).



36
37
38
39
40
 
# File 'lib/dispatch/job.rb', line 36

def value(queue = Dispatch::Queue.concurrent, &block)
  return group.notify(queue) { block.call(result) } if not block.nil?
  group.wait
  return result
end