[rabbitmq-discuss] Old consumer tags delivered after re-subscribing to a queue

Thu Apr 1 16:05:25 BST 2010

I'm facing the following issue with RabbitMQ. 
After unsubscribing from a queue, following "recover" calls will still redeliver unack'd messages through the channel. Even worse, after resubscribing to the queue, the recovered messages are still sent with the old consumer tag.

I'm using the amqp ruby library to consume the queues. Here's a small snippet that reproduces the issue. When I execute this, I always get an error similar to: Basic.Deliver for invalid consumer tag: myqueue-396735623077.

==========

require 'rubygems'
require 'amqp'
require 'mq'

Thread.new { EM.run}

AMQP.start(:host => 'localhost', :logging => false)

MQ.error do |msg|
  puts "ERROR: #{msg}"
end

@exchange = MQ.direct("myexchange")  
@queue = MQ.queue("myqueue").bind(@exchange)

def subscribe
  @queue.subscribe(:ack => true) do |msg|
    puts "Message received: #{msg}"
  end
end

subscribe ; sleep 0.2
@exchange.publish Kernel.rand(1E10).to_s ; sleep 0.2
@queue.unsubscribe ; sleep 0.2
subscribe ; sleep 0.2
MQ.recover ; sleep 0.2

==========

The reason I'm doing this is because I want to accommodate to availability of the target services. So, whenever any component is temporarily down or full, I want to stop receiving messages from the queue. Maybe is there any other way to achieve this?

I'm using RabbitMQ 1.7.2 and the ruby library is amqp 0.6.6.

Thanks,
- Juan