More work on queue, binding and consumer recovery

To recover consumers we will have to make sure Bunny::Consumer instance
always have a reference to their respective queue object. The reason for
that is that when a server-named queue is recovered, consumer needs to
use the new name.

lib/bunny/channel.rb

@@ -610,25 +610,34 @@ def recover_from_network_failure
 
       recover_prefetch_setting
       recover_exchanges
+      # this includes recovering bindings
       recover_queues
+      recover_consumers
     end
 
     def recover_prefetch_setting
       basic_qos(@prefetch_count) if @prefetch_count
     end
 
     def recover_exchanges
-      @exchanges.each do |_, x|
+      @exchanges.values.dup.each do |x|
         x.recover_from_network_failure
       end
     end
 
     def recover_queues
-      @queues.each do |_, q|
+      @queues.values.dup.each do |q|
         q.recover_from_network_failure
       end
     end
 
+    def recover_consumers
+      @consumers.values.dup.each do |c|
+        c.recover_from_network_failure
+      end
+    end
+
+
 
     #
     # Implementation
@@ -785,6 +794,10 @@ def deregister_queue(queue)
       @queues.delete(queue.name)
     end
 
+    def deregister_queue_named(name)
+      @queues.delete(name)
+    end
+
     def register_queue(queue)
       @queues[queue.name] = queue
     end

lib/bunny/consumer.rb

@@ -58,5 +58,13 @@ def cancel
     def inspect
       "#<#{self.class.name}:#{object_id} @channel_id=#{@channel.number} @queue=#{self.queue_name}> @consumer_tag=#{@consumer_tag} @exclusive=#{@exclusive} @no_ack=#{@no_ack}>"
     end
+
+    #
+    # Recovery
+    #
+
+    def recover_from_network_failure
+      # TODO
+    end
   end
 end

lib/bunny/queue.rb

@@ -76,7 +76,7 @@ def bind(exchange, opts = {})
 
       # store bindings for automatic recovery. We need to be very careful to
       # not cause an infinite rebinding loop here when we recover. MK.
-      binding = { :exchange => exchange_name, :routing_key => (opts[:routing_key] || opts[:key]), :arguments => arguments }
+      binding = { :exchange => exchange_name, :routing_key => (opts[:routing_key] || opts[:key]), :arguments => opts[:arguments] }
       @bindings.push(binding) unless @bindings.include?(binding)
 
       self
@@ -195,8 +195,33 @@ def consumer_count
       s[:consumer_count]
     end
 
+    #
+    # Recovery
+    #
+
     def recover_from_network_failure
-      puts "Recovering queue #{@name} from network failure"
+      # puts "Recovering queue #{@name} from network failure"
+
+      if self.server_named?
+        old_name = @name.dup
+        @name    = AMQ::Protocol::EMPTY_STRING
+
+        @channel.deregister_queue_named(old_name)
+      end
+
+      declare!
+      begin
+        @channel.register_queue(self)
+      rescue Exception => e
+        puts "Caught #{e.inspect} while registering #{@name}!"
+      end
+      recover_bindings
+    end
+
+    def recover_bindings
+      @bindings.each do |b|
+        self.bind(b[:exchange], b)
+      end
     end