Rabbit

gistfile1.txt

require 'bunny'

connection = Bunny.new
connection.start

channel  = connection.create_channel
channel.prefetch(3)

exchange = channel.direct("moar")

cnt = 0
messages = []

queue = channel.queue("", :durable => true)
queue.purge #just for this test

queue.bind(exchange).subscribe(:ack => true, :block => false) do |delivery_info, metadata, payload|
 # Display message count
  puts queue.message_count

  puts "Got the #{payload}"
  messages << payload  
  cnt +=1
  if cnt  >= 3 || queue.message_count == 0
    messages.each do |message|
      puts "Added #{message} to transaction"
    end
    puts "Batched #{messages.size}:"
    channel.acknowledge(delivery_info.delivery_tag, true) #true acknowledges all 3 messages
    cnt = 0
    messages = []
  end
end

20.times do |t| 
  exchange.publish("Hello #{t}!")
end

sleep 1.0
connection.close

#The problem with this method is that the 10th message gets stuck waiting in a queue.