I did not find a way to do it easily using Bunny, and what I propose here blocks without a timeout. But it does support retrieving one message per call semantics. Given that Bunny internally uses a thread-pool to receive messages, I figured that a simpler way might be to use a blocking queue such as Ruby's Queue
class for transporting the message from Bunny's thread-pool to the calling thread. Something like the following:
# Set up your internal queue somewhere (in your class's initialize maybe?)
@internal_queue = Queue.new
# In the main thread that needs to block
...
# the call to subscribe is non-blocking
queue.subscribe do |delivery_info, properties, payload|
@internal_queue.enq(payload) # this runs inside Bunny's pool
end
# the call to deq is blocking
response = @internal_queue.deq # this blocks the main thread till a
# message is pushed to the internal_q
You could maintain one @internal_queue for each AMQP channel you need to listen on. You could factor these parts out into separate methods and make a neat blocking API that returns you one message at a time.
I later created a TimedWaitableQueue class wrapping a simple Array extended with monitor MonitorMixin, and then using mutex + condition variable semantics. That allowed blocking on a dequeue call with timeouts.