1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
require "thread"
module Bunny
module Concurrent
# Continuation queue implementation for MRI and Rubinius
#
# @private
class ContinuationQueue
def initialize
@q = []
@lock = ::Mutex.new
@cond = ::ConditionVariable.new
end
def push(item)
@lock.synchronize do
@q.push(item)
@cond.signal
end
end
alias << push
def pop
poll
end
def poll(timeout_in_ms = nil)
timeout_in_sec = timeout_in_ms ? timeout_in_ms / 1000.0 : nil
@lock.synchronize do
started_at = Bunny::Timestamp.monotonic
while @q.empty?
wait = !(timeout_in_sec.nil?)
@cond.wait(@lock, timeout_in_sec)
if wait
ended_at = Bunny::Timestamp.monotonic
elapsed = ended_at - started_at
raise ::Timeout::Error if (elapsed > timeout_in_sec)
end
end
item = @q.shift
item
end
end
def clear
@lock.synchronize do
@q.clear
end
end
def empty?
@q.empty?
end
def size
@q.size
end
alias length size
end
end
end
|
