Class: Async::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/async/queue.rb

Overview

A queue which allows items to be processed in order.

It has a compatible interface with Notification and Condition, except that it’s multi-value.

Direct Known Subclasses

LimitedQueue

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(parent: nil, available: Notification.new) ⇒ Queue

Create a new queue.



21
22
23
24
25
 
# File 'lib/async/queue.rb', line 21

def initialize(parent: nil, available: Notification.new)
  @items = []
  @parent = parent
  @available = available
end

Instance Attribute Details

#items ⇒ Object (readonly)

Returns the value of attribute items.



28
29
30
 
# File 'lib/async/queue.rb', line 28

def items
  @items
end

#The items in the queue.(items) ⇒ Object (readonly) 



28
 
# File 'lib/async/queue.rb', line 28

attr :items

Instance Method Details

#<<(item) ⇒ Object

Compatibility with Queue#push.



48
49
50
 
# File 'lib/async/queue.rb', line 48

def <<(item)
  self.push(item)
end

#async(parent: (@parent or Task.current), **options, &block) ⇒ Object

Process each item in the queue.



81
82
83
84
85
 
# File 'lib/async/queue.rb', line 81

def async(parent: (@parent or Task.current), **options, &block)
  while item = self.dequeue
    parent.async(item, **options, &block)
  end
end

#dequeue ⇒ Object

Remove and return the next item from the queue.



60
61
62
63
64
65
66
 
# File 'lib/async/queue.rb', line 60

def dequeue
  while @items.empty?
    @available.wait
  end
  
  @items.shift
end

#each ⇒ Object

Enumerate each item in the queue.



88
89
90
91
92
 
# File 'lib/async/queue.rb', line 88

def each
  while item = self.dequeue
    yield item
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


36
37
38
 
# File 'lib/async/queue.rb', line 36

def empty?
  @items.empty?
end

#enqueue(*items) ⇒ Object

Add multiple items to the queue.



53
54
55
56
57
 
# File 'lib/async/queue.rb', line 53

def enqueue(*items)
  @items.concat(items)
  
  @available.signal unless self.empty?
end

#pop ⇒ Object

Compatibility with Queue#pop.



69
70
71
 
# File 'lib/async/queue.rb', line 69

def pop
  self.dequeue
end

#push(item) ⇒ Object

Add an item to the queue.



41
42
43
44
45
 
# File 'lib/async/queue.rb', line 41

def push(item)
  @items << item
  
  @available.signal unless self.empty?
end

#signal(value = nil) ⇒ Object

Signal the queue with a value, the same as #enqueue.



95
96
97
 
# File 'lib/async/queue.rb', line 95

def signal(value = nil)
  self.enqueue(value)
end

#size ⇒ Object 



31
32
33
 
# File 'lib/async/queue.rb', line 31

def size
  @items.size
end

#wait ⇒ Object

Wait for an item to be available, the same as #dequeue.



100
101
102
 
# File 'lib/async/queue.rb', line 100

def wait
  self.dequeue
end