Class: PgPipeline::BoundedQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/pg_pipeline/bounded_queue.rb

Instance Method Summary collapse

Constructor Details

#initialize(limit) ⇒ BoundedQueue

Returns a new instance of BoundedQueue.



8
9
10
11
12
13
14
15
16
17
18
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 8

def initialize(limit)
  @limit = Integer(limit)
  raise ArgumentError, "limit must be >= 1" if @limit < 1
rescue ArgumentError, TypeError
  raise ArgumentError, "limit must be an integer >= 1"
else
  @items, @consumers, @producers = [], [], []

  @closed = false
  @close_error = nil
end

Instance Method Details

#close(error) ⇒ Object 



48
49
50
51
52
53
54
55
56
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 48

def close(error)
  return if @closed

  @closed = true
  @close_error = error
  wake_all(@consumers)
  wake_all(@producers)
  nil
end

#dequeue ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 34

def dequeue
  while true
    unless @items.empty?
      item = @items.shift
      wake_one(@producers)
      return item
    end

    return nil if @closed

    wait_on(@consumers)
  end
end

#drain ⇒ Object 



63
64
65
66
67
68
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 63

def drain
  items = @items
  @items = []
  wake_all(@producers)
  items
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


58
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 58

def empty? = @items.empty?

#enqueue(item) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 20

def enqueue(item)
  while true
    raise_close_error if @closed

    if @items.size < @limit
      @items << item
      wake_one(@consumers)
      return item
    end

    wait_on(@producers)
  end
end

#size ⇒ Object 



59
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 59

def size = @items.size

#waiting_consumers ⇒ Object 



61
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 61

def waiting_consumers = @consumers.size

#waiting_producers ⇒ Object 



60
 
# File 'lib/pg_pipeline/bounded_queue.rb', line 60

def waiting_producers = @producers.size