Ruby: adopt Bunny 3.1.0, port tutorial 7 (publisher confirms)

Now that Bunny `3.x.` provides publisher
confirm tracking and backpressure just like
.NET, Swift 6 clients do.

Author: michaelklishin

ruby/Gemfile

@@ -1,5 +1,5 @@
 source 'https://rubygems.org'
 
-ruby ">= 2.5.0"
+ruby ">= 3.1.0"
 
-gem "bunny", ">= 2.14.2", " < 3.0"
+gem "bunny", "~> 3.1.0"

ruby/Gemfile.lock

@@ -1,18 +1,24 @@
 GEM
   remote: https://rubygems.org/
   specs:
-    amq-protocol (2.3.0)
-    bunny (2.14.2)
-      amq-protocol (~> 2.3, >= 2.3.0)
+    amq-protocol (2.7.0)
+    bunny (3.1.0)
+      amq-protocol (~> 2.7)
+      logger (~> 1, >= 1.7)
+      sorted_set (~> 1, >= 1.0.2)
+    logger (1.7.0)
+    rbtree (0.4.6)
+    sorted_set (1.1.0)
+      rbtree
 
 PLATFORMS
   ruby
 
 DEPENDENCIES
-  bunny (>= 2.14.2, < 3.0)
+  bunny (~> 3.1.0)
 
 RUBY VERSION
-   ruby 2.5.1p57
+   ruby 3.1.0
 
 BUNDLED WITH
-   1.17.2
+   2.5.23

ruby/README.md

@@ -5,7 +5,7 @@ Here you can find Ruby code examples from
 
 ## Requirements
 
-These tutorials require Ruby 2.2 or later, [Bundler](https://bundler.io/) and [Bunny](http://rubybunny.info) to be installed.
+These tutorials require Ruby 3.1.0 or later, [Bundler](https://bundler.io/) and [Bunny](http://rubybunny.info) to be installed.
 
 To install Bunny with Bundler, do
 
@@ -58,4 +58,10 @@ bundle exec ruby rpc_server.rb
 bundle exec ruby rpc_client.rb
 ```
 
+[Tutorial seven: Publisher Confirms](https://www.rabbitmq.com/tutorials/tutorial-seven-ruby.html)
+
+``` sh
+bundle exec ruby publisher_confirms.rb
+```
+
 To learn more, see [Bunny documentation](http://rubybunny.info).

ruby/publisher_confirms.rb

@@ -0,0 +1,102 @@
+#!/usr/bin/env ruby
+require 'bunny'
+
+MESSAGE_COUNT = 50_000
+
+def publish_messages_individually(connection)
+  channel = connection.create_channel
+  queue = channel.queue('', exclusive: true)
+
+  channel.confirm_select(tracking: true)
+
+  start_time = Time.now
+  MESSAGE_COUNT.times do |i|
+    body = i.to_s
+    channel.basic_publish(body, '', queue.name)
+  end
+  
+  # Wait for any remaining confirmations
+  channel.wait_for_confirms
+  end_time = Time.now
+
+  puts "Published #{MESSAGE_COUNT} messages individually in #{((end_time - start_time) * 1000).to_i} ms"
+ensure
+  channel.close if channel && channel.open?
+end
+
+def publish_messages_in_batch(connection)
+  channel = connection.create_channel
+  queue = channel.queue('', exclusive: true)
+
+  channel.confirm_select(tracking: true)
+
+  batch_size = 1000
+  start_time = Time.now
+
+  (0...MESSAGE_COUNT).each_slice(batch_size) do |batch|
+    messages = batch.map { |i| i.to_s }
+    channel.basic_publish_batch(messages, '', queue.name)
+  end
+  
+  # Wait for any remaining confirmations
+  channel.wait_for_confirms
+  end_time = Time.now
+
+  puts "Published #{MESSAGE_COUNT} messages in batch in #{((end_time - start_time) * 1000).to_i} ms"
+ensure
+  channel.close if channel && channel.open?
+end
+
+def handle_publish_confirms_asynchronously(connection)
+  channel = connection.create_channel
+  queue = channel.queue('', exclusive: true)
+
+  outstanding_confirms = {}
+  # A mutex is necessary because the confirm callbacks are executed in a separate thread
+  confirms_mutex = Mutex.new
+
+  channel.confirm_select do |delivery_tag, multiple, nack|
+    confirms_mutex.synchronize do
+      if multiple
+        outstanding_confirms.reject! { |k, _| k <= delivery_tag }
+      else
+        outstanding_confirms.delete(delivery_tag)
+      end
+    end
+    if nack
+      puts "Message with delivery tag #{delivery_tag} was nacked!"
+    end
+  end
+
+  start_time = Time.now
+  MESSAGE_COUNT.times do |i|
+    body = i.to_s
+    seq_no = channel.next_publish_seq_no
+    confirms_mutex.synchronize do
+      outstanding_confirms[seq_no] = body
+    end
+    channel.basic_publish(body, '', queue.name)
+  end
+
+  # Wait for any remaining confirmations
+  channel.wait_for_confirms
+  end_time = Time.now
+
+  puts "Published #{MESSAGE_COUNT} messages and handled confirms asynchronously in #{((end_time - start_time) * 1000).to_i} ms"
+ensure
+  channel.close if channel && channel.open?
+end
+
+begin
+  connection = Bunny.new
+  connection.start
+
+  publish_messages_individually(connection)
+  publish_messages_in_batch(connection)
+  handle_publish_confirms_asynchronously(connection)
+rescue Interrupt => _
+  connection.close
+  exit(0)
+ensure
+  connection.close if connection
+end