Introduce a few helpers for ruby-amqp/hutch#414

Author: michaelklishin

AGENTS.md

@@ -14,6 +14,30 @@ This library targets Ruby 3.0 and later versions.
 
  * Only add very important comments, both in tests and in the implementation
 
+## Change Log
+
+If asked to perform change log updates, consult and modify `ChangeLog.md` and stick to its
+existing writing style.
+
+
+## Releases
+
+### How to Roll (Produce) a New Release
+
+Suppose the current development version in `ChangeLog.md` has
+a `## Changes between Bunny X.Y.0 and X.(Y+1).0 (in development)` section at the top.
+
+To produce a new release:
+
+ 1. Update `ChangeLog.md`: replace `(in development)` with today's date, e.g. `(Mar 30, 2026)`. Make sure all notable changes since the previous release are listed
+ 2. Update the version in `lib/bunny/version.rb` to match (remove any `.pre` suffix)
+ 3. Commit with the message `X.(Y+1).0` (just the version number, nothing else)
+ 4. Tag the commit: `git tag vX.(Y+1).0`
+ 5. Bump the dev version: add a new `## Changes between Bunny X.(Y+1).0 and X.(Y+2).0 (in development)` section to `ChangeLog.md` with `No changes yet.` underneath, and update `lib/bunny/version.rb` to the next dev version with a `.pre` suffix
+ 6. Commit with the message `Bump dev version`
+ 7. Push: `git push && git push --tags`
+
+
 ## Git Instructions
 
  * Never add yourself to the list of commit co-authors

ChangeLog.md

@@ -158,6 +158,24 @@ Several optimizations to reduce overhead in the consumer delivery hot path:
  * Frame header buffer reuse: the transport layer now reuses a buffer when reading
    frame headers, reducing per-frame allocations
 
+### `Channel#reopen`
+
+A new method that reopens a channel after a server-initiated closure
+(e.g. due to a consumer delivery acknowledgement timeout or an unknown delivery tag).
+The channel is reopened on the same connection, reusing its original channel id,
+and its prefetch, confirm, and transactional settings are recovered.
+
+### `Session#recover_channel_topology`
+
+Recovers topology (exchanges, queues, bindings, consumers) for a single channel.
+Intended for use after `Channel#reopen`.
+
+### `amq-protocol` Bumped to `2.6.0` (or Later)
+
+Bunny now requires `amq-protocol` `2.6.0` or later for the `Channel::Close`
+predicate methods (`#unknown_delivery_tag?`, `#delivery_ack_timeout?`, `#message_too_large?`)
+that Bunny used to reinvent (with regular expression matches on `reply_text`)
+
 ### Limit Hostname Resolution Time
 
 Bunny now configures its TCP socket to limit the hostname resolution time,

Gemfile

@@ -41,14 +41,15 @@ gemspec
 
 # Use local clones if possible.
 # If you want to use your local copy, just symlink it to vendor.
-def custom_gem(name, options = Hash.new)
+def custom_gem(name, *args)
+  options = args.last.is_a?(Hash) ? args.pop : {}
   local_path = File.expand_path("../vendor/#{name}", __FILE__)
   if File.exist?(local_path)
     puts "Using #{name} from #{local_path}..."
     gem name, options.merge(path: local_path).delete_if { |key, _| [:git, :branch].include?(key) }
   else
-    gem name, options
+    gem name, *args, **options
   end
 end
 
-custom_gem "amq-protocol", "~> 2.5.0"
+custom_gem "amq-protocol", "~> 2.6"

bunny.gemspec

@@ -28,7 +28,7 @@ Gem::Specification.new do |s|
   s.email = ["michael.s.klishin@gmail.com"]
 
   # Dependencies
-  s.add_runtime_dependency 'amq-protocol', '~> 2.5.1'
+  s.add_runtime_dependency 'amq-protocol', '~> 2.6'
   s.add_runtime_dependency 'logger', '~> 1', '>= 1.7'
   s.add_runtime_dependency 'sorted_set', '~> 1', '>= 1.0.2'
 

lib/bunny/channel.rb

@@ -298,6 +298,35 @@ def close
       maybe_kill_consumer_work_pool!
     end
 
+    # Reopens a channel that was closed by the server (e.g. due to a consumer
+    # delivery acknowledgement timeout). The channel is reopened on the same
+    # connection, reusing its original channel id, and its prefetch, confirm,
+    # and transactional settings are recovered.
+    #
+    # This does NOT recover topology (queues, exchanges, bindings, consumers).
+    # Use {Bunny::Session#recover_channel_topology} for that.
+    #
+    # @return [Bunny::Channel] self
+    # @see Bunny::Session#recover_channel_topology
+    # @api public
+    def reopen
+      raise "Cannot reopen a channel that is not closed" unless closed?
+
+      existing = @connection.synchronised_find_channel(@id)
+      if existing && existing != self
+        raise "Channel id #{@id} has been reassigned to another channel"
+      end
+
+      @work_pool = ConsumerWorkPool.new(@work_pool.size, @work_pool.abort_on_exception)
+      @work_pool.start
+
+      open
+
+      recover_from_network_failure
+
+      self
+    end
+
     # @return [Boolean] true if this channel is open, false otherwise
     # @api public
     def open?
@@ -2175,14 +2204,7 @@ def handle_method(method)
 
     # @private
     def channel_level_exception_after_operation_that_has_no_response?(method)
-      re1 = /unknown delivery tag/
-      re2 = /delivery acknowledgement on channel \d+ timed out/
-      re3 = /larger than configured max size/
-
-      res = [re1, re2, re3]
-      desc = method.reply_text
-
-      method.reply_code == 406 && (res.any? { |re| desc =~ re })
+      method.unknown_delivery_tag? || method.delivery_ack_timeout? || method.message_too_large?
     end
 
     # @private

lib/bunny/session.rb

@@ -627,6 +627,17 @@ def after_recovery_attempts_exhausted(&block)
       @recovery_attempts_exhausted = block
     end
 
+    # Recovers topology (exchanges, queues, bindings, consumers) for a single channel.
+    # Intended for use after {Bunny::Channel#reopen} to restore the channel to a
+    # working state with its original consumers.
+    #
+    # @param [Bunny::Channel] ch Channel whose topology should be recovered
+    # @api public
+    def recover_channel_topology(ch)
+      filter = Proc.new { |entity| entity.channel == ch }
+      recover_topology_with(filter)
+    end
+
     #
     # Implementation
     #
@@ -1045,46 +1056,63 @@ def recover_channels
     # @private
     def recover_topology
       @logger.debug "Will recover topology now"
-      # The recovery sequence is the following:
-      # 1. Recover exchanges
-      @logger.debug "Will recover recorded exchanges"
-      @topology_registry.filtered_exchanges.reject { |x| x.predeclared? }.each do |rx|
+      recover_topology_with
+    end
+
+    # @private
+    def recover_topology_with(filter = nil)
+      exchanges = @topology_registry.filtered_exchanges.reject(&:predeclared?)
+      queues = @topology_registry.filtered_queues
+      queue_bindings = @topology_registry.filtered_queue_bindings
+      exchange_bindings = @topology_registry.filtered_exchange_bindings
+      consumers = @topology_registry.filtered_consumers
+
+      if filter
+        exchanges = exchanges.select(&filter)
+        queues = queues.select(&filter)
+        queue_bindings = queue_bindings.select(&filter)
+        exchange_bindings = exchange_bindings.select(&filter)
+        consumers = consumers.select(&filter)
+      end
+
+      @logger.debug { "Will recover #{exchanges.size} exchange(s)" }
+      exchanges.each do |x|
         begin
-          recover_exchange(rx)
+          recover_exchange(x)
         rescue Exception => e
-          @logger.error "Caught an exception while re-declaring exchange #{rx.name}: #{e.inspect}"
+          @logger.error "Caught an exception while recovering exchange #{x.name}: #{e.inspect}"
         end
       end
-      # 2. Recover queues
-      @logger.debug "Will recover recorded queues"
-      @topology_registry.filtered_queues.each do |rq|
+
+      @logger.debug { "Will recover #{queues.size} queue(s)" }
+      queues.each do |q|
         begin
-          recover_queue(rq)
+          recover_queue(q)
         rescue Exception => e
-          @logger.error "Caught an exception while re-declaring queue #{rq.name}: #{e.inspect}"
+          @logger.error "Caught an exception while recovering queue #{q.name}: #{e.inspect}"
         end
       end
-      # 3. Recover bindings
-      @logger.debug "Will recover recorded bindings"
-      @topology_registry.filtered_queue_bindings.each do |rb|
+
+      @logger.debug { "Will recover #{queue_bindings.size + exchange_bindings.size} binding(s)" }
+      queue_bindings.each do |b|
         begin
-          recover_queue_binding(rb)
+          recover_queue_binding(b)
         rescue Exception => e
-          @logger.error "Caught an exception while re-declaring a binding of queue #{rb.destination}: #{e.inspect}"
+          @logger.error "Caught an exception while recovering a binding of queue #{b.destination}: #{e.inspect}"
         end
       end
-      @topology_registry.filtered_exchange_bindings.each do |rb|
+
+      exchange_bindings.each do |b|
         begin
-          recover_exchange_binding(rb)
+          recover_exchange_binding(b)
         rescue Exception => e
-          @logger.error "Caught an exception while re-declaring a binding of exchange #{rb.source}: #{e.inspect}"
+          @logger.error "Caught an exception while recovering a binding of exchange #{b.source}: #{e.inspect}"
         end
       end
 
-      # 4. Recover consumers
-      @logger.debug "Will recover recorded consumers"
-      @topology_registry.filtered_consumers.each do |rc|
-        recover_consumer(rc)
+      @logger.debug { "Will recover #{consumers.size} consumer(s)" }
+      consumers.each do |c|
+        recover_consumer(c)
       end
     end
 

spec/higher_level_api/integration/channel_reopen_spec.rb

@@ -0,0 +1,94 @@
+require "spec_helper"
+
+describe Bunny::Channel, "#reopen" do
+  let(:connection) do
+    c = Bunny.new(username: "bunny_gem", password: "bunny_password", vhost: "bunny_testbed")
+    c.start
+    c
+  end
+
+  after :each do
+    connection.close if connection.open?
+  end
+
+  it "reopens a channel after a server-initiated closure" do
+    ch = connection.create_channel
+    q  = ch.queue("bunny.test.channel-reopen.#{rand}", exclusive: true)
+
+    ch.on_error do |closed_ch, amq_close|
+      @channel_close = amq_close
+    end
+
+    # ack with an invalid tag to force a 406 channel closure
+    ch.ack(82, false)
+    sleep 0.25
+
+    expect(@channel_close.reply_code).to eq AMQ::Protocol::PreconditionFailed::VALUE
+    expect(@channel_close).to be_unknown_delivery_tag
+    expect(@channel_close).not_to be_delivery_ack_timeout
+    expect(ch).to be_closed
+
+    ch.reopen
+    expect(ch).to be_open
+
+    # the channel should be functional again
+    q2 = ch.queue("bunny.test.channel-reopen.after.#{rand}", exclusive: true)
+    ch.default_exchange.publish("hello", routing_key: q2.name)
+    sleep 0.25
+    expect(q2.message_count).to eq 1
+  end
+
+  it "recovers prefetch setting after reopen" do
+    ch = connection.create_channel
+    ch.prefetch(5)
+
+    ch.on_error do |_, _| end
+
+    ch.ack(82, false)
+    sleep 0.25
+    expect(ch).to be_closed
+
+    ch.reopen
+    expect(ch).to be_open
+    expect(ch.prefetch_count).to eq 5
+  end
+
+  it "raises when called on a channel that is still open" do
+    ch = connection.create_channel
+    expect { ch.reopen }.to raise_error(RuntimeError, /not closed/)
+  end
+
+  context "with recover_channel_topology" do
+    it "re-registers consumers that receive new messages" do
+      ch = connection.create_channel
+      ch.prefetch(1)
+      q  = ch.queue("bunny.test.channel-reopen.topology.#{rand}", exclusive: true)
+      x  = ch.default_exchange
+
+      delivered = []
+      q.subscribe(manual_ack: true) do |di, _props, payload|
+        delivered << payload
+        ch.ack(di.delivery_tag)
+      end
+
+      x.publish("before", routing_key: q.name)
+      sleep 0.5
+      expect(delivered).to eq ["before"]
+
+      ch.on_error do |_, _| end
+
+      # force channel closure
+      ch.ack(999, false)
+      sleep 0.25
+      expect(ch).to be_closed
+
+      ch.reopen
+      connection.recover_channel_topology(ch)
+      sleep 0.25
+
+      x.publish("after", routing_key: q.name)
+      sleep 0.5
+      expect(delivered).to include("after")
+    end
+  end
+end