Class: Chronos::Internal::BoundedQueue

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

Overview

Thread-safe queue with fixed capacity and non-blocking producer behavior.

Examples:

queue.push(event) #=> true or false

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(capacity) ⇒ BoundedQueue

Returns a new instance of BoundedQueue.

Raises:

  • (ArgumentError) 


17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/chronos/internal/bounded_queue.rb', line 17

def initialize(capacity)
  raise ArgumentError, "capacity must be positive" unless capacity.is_a?(Integer) && capacity > 0

  @capacity = capacity
  @items = []
  @accepted = 0
  @dropped = 0
  @closed = false
  @mutex = Mutex.new
  @condition = ConditionVariable.new
end

Instance Attribute Details

#capacity ⇒ Object (readonly)

Returns the value of attribute capacity.



15
16
17
 
# File 'lib/chronos/internal/bounded_queue.rb', line 15

def capacity
  @capacity
end

Instance Method Details

#close ⇒ Object 



52
53
54
55
56
57
58
 
# File 'lib/chronos/internal/bounded_queue.rb', line 52

def close
  @mutex.synchronize do
    @closed = true
    @condition.broadcast
  end
  true
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


60
61
62
 
# File 'lib/chronos/internal/bounded_queue.rb', line 60

def closed?
  @mutex.synchronize { @closed }
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


64
65
66
 
# File 'lib/chronos/internal/bounded_queue.rb', line 64

def empty?
  @mutex.synchronize { @items.empty? }
end

#pop(timeout = nil) ⇒ Object 



44
45
46
47
48
49
50
 
# File 'lib/chronos/internal/bounded_queue.rb', line 44

def pop(timeout = nil)
  @mutex.synchronize do
    @condition.wait(@mutex, timeout) while @items.empty? && !@closed && timeout.nil?
    @condition.wait(@mutex, timeout) if @items.empty? && !@closed && timeout
    @items.shift
  end
end

#push(item) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/chronos/internal/bounded_queue.rb', line 29

def push(item)
  @mutex.synchronize do
    return false if @closed
    if @items.length >= @capacity
      @dropped += 1
      return false
    end

    @items << item
    @accepted += 1
    @condition.signal
    true
  end
end

#size ⇒ Object 



68
69
70
 
# File 'lib/chronos/internal/bounded_queue.rb', line 68

def size
  @mutex.synchronize { @items.size }
end

#stats ⇒ Object 



72
73
74
75
76
 
# File 'lib/chronos/internal/bounded_queue.rb', line 72

def stats
  @mutex.synchronize do
    {:size => @items.size, :capacity => capacity, :accepted => @accepted, :dropped => @dropped}
  end
end