class Bunny::Concurrent::ContinuationQueue
Continuation queue implementation for MRI and Rubinius
@private
Public Class Methods
Source
# File lib/bunny/concurrent/continuation_queue.rb, line 11 def initialize @q = [] @lock = ::Mutex.new @cond = ::ConditionVariable.new end
Public Instance Methods
Source
# File lib/bunny/concurrent/continuation_queue.rb, line 49 def clear @lock.synchronize do @q.clear end end
Source
# File lib/bunny/concurrent/continuation_queue.rb, line 55 def empty? @q.empty? end
Source
# File lib/bunny/concurrent/continuation_queue.rb, line 29 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
Source
# File lib/bunny/concurrent/continuation_queue.rb, line 17 def push(item) @lock.synchronize do @q.push(item) @cond.signal end end
Also aliased as: <<