From 6a9053b3fa09004a4107549861e223a94b8bed58 Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Mon, 17 Mar 2025 15:08:37 +0100
Subject: [PATCH 01/10] Start to sketch experiment

---
 lib/beetle/no_deduplication_store.rb | 64 ++++++++++++++++++++++++++++
 test/beetle/message_test.rb          | 25 +++++++++++
 2 files changed, 89 insertions(+)
 create mode 100644 lib/beetle/no_deduplication_store.rb

diff --git a/lib/beetle/no_deduplication_store.rb b/lib/beetle/no_deduplication_store.rb
new file mode 100644
index 00000000..866c4708
--- /dev/null
+++ b/lib/beetle/no_deduplication_store.rb
@@ -0,0 +1,64 @@
+module Beetle
+  class NoDeduplicationStore < DeduplicationStore
+    include Logging
+
+    def initialize(config = Beetle.config)
+      super(config)
+      @data = {}
+    end
+
+    def redis
+      raise NotImplementedError
+    end
+
+    def set(msg_id, suffix, value)
+      @data[key(msg_id, suffix)] = value
+    end
+
+    def setnx(msg_id, suffix, value)
+      @data[key(msg_id, suffix)] ||= value
+    end
+
+    def setnx_completed!(_msg_id)
+      true
+    end
+
+    def mset(msg_id, values)
+      values.each do |k, v|
+        set(msg_id, k, v)
+      end
+    end
+
+    def msetnx(msg_id, values)
+      values.map { |k, _v| key(msg_id, k) }
+
+      return unless values.keys.intersection(@data.keys).empty?
+      mset(msg_id, values)
+    end
+
+    def incr(msg_id, suffix)
+      @data[key(msg_id, suffix)] ||= 0
+      @data[key(msg_id, suffix)] += 1
+    end
+
+    def get(msg_id, suffix)
+      @data[key(msg_id, suffix)]
+    end
+
+    def del(msg_id, suffix)
+      @data[key(msg_id, suffix)] = nil
+    end
+
+    def del_keys(msg_id)
+      keys(msg_id).each do |key|
+        @data[key] = nil
+      end
+    end
+
+    def flushdb
+      @data = {}
+    end
+
+    def with_failover; end
+  end
+end
diff --git a/test/beetle/message_test.rb b/test/beetle/message_test.rb
index e0aab1bd..38cfde88 100644
--- a/test/beetle/message_test.rb
+++ b/test/beetle/message_test.rb
@@ -723,6 +723,8 @@ class HandlerTimeoutTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
       @store.flushdb
+      @no_store = NoDeduplicationStore.new
+      @no_store.flushdb
     end
 
     test "a handler running longer than the specified timeout should be aborted (when given a float timeout number)" do
@@ -735,6 +737,18 @@ def setup
       assert_equal RC::ExceptionsLimitReached, result
     end
 
+    test "[NODEDUP] a handler running longer than the specified timeout should be aborted (when given a float timeout number)" do
+      header = header_with_params({})
+      header.expects(:ack)
+      message = Message.new("somequeue", header, 'foo', :timeout => 0.1, :attempts => 2, :store => @no_store)
+      action = lambda{|*args| while true; end}
+      handler = Handler.create(action)
+      result = message.process(handler)
+      assert_equal RC::ExceptionsLimitReached, result
+    end
+
+
+
     test "a handler running longer than the specified timeout should be aborted (when using active_support seconds)" do
       header = header_with_params({})
       header.expects(:ack)
@@ -745,6 +759,17 @@ def setup
       assert_equal RC::ExceptionsLimitReached, result
     end
 
+    test "[NODEDUP] a handler running longer than the specified timeout should be aborted (when using active_support seconds)" do
+      header = header_with_params({})
+      header.expects(:ack)
+      message = Message.new("somequeue", header, 'foo', :timeout => 1.seconds, :attempts => 2, :store => @no_store)
+      action = lambda{|*args| while true; end}
+      handler = Handler.create(action)
+      result = message.process(handler)
+      assert_equal RC::ExceptionsLimitReached, result
+    end
+
+
   end
 
   class MySQLFailoverTest < Minitest::Test

From 54a7145f1052563b3184f9ac9d39f3697627c42f Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Mon, 17 Mar 2025 15:31:09 +0100
Subject: [PATCH 02/10] better tests

---
 test/beetle/message_test.rb | 29 +++++------------------------
 1 file changed, 5 insertions(+), 24 deletions(-)

diff --git a/test/beetle/message_test.rb b/test/beetle/message_test.rb
index 38cfde88..9d5a4c39 100644
--- a/test/beetle/message_test.rb
+++ b/test/beetle/message_test.rb
@@ -723,8 +723,6 @@ class HandlerTimeoutTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
       @store.flushdb
-      @no_store = NoDeduplicationStore.new
-      @no_store.flushdb
     end
 
     test "a handler running longer than the specified timeout should be aborted (when given a float timeout number)" do
@@ -737,18 +735,6 @@ def setup
       assert_equal RC::ExceptionsLimitReached, result
     end
 
-    test "[NODEDUP] a handler running longer than the specified timeout should be aborted (when given a float timeout number)" do
-      header = header_with_params({})
-      header.expects(:ack)
-      message = Message.new("somequeue", header, 'foo', :timeout => 0.1, :attempts => 2, :store => @no_store)
-      action = lambda{|*args| while true; end}
-      handler = Handler.create(action)
-      result = message.process(handler)
-      assert_equal RC::ExceptionsLimitReached, result
-    end
-
-
-
     test "a handler running longer than the specified timeout should be aborted (when using active_support seconds)" do
       header = header_with_params({})
       header.expects(:ack)
@@ -758,18 +744,13 @@ def setup
       result = message.process(handler)
       assert_equal RC::ExceptionsLimitReached, result
     end
+  end
 
-    test "[NODEDUP] a handler running longer than the specified timeout should be aborted (when using active_support seconds)" do
-      header = header_with_params({})
-      header.expects(:ack)
-      message = Message.new("somequeue", header, 'foo', :timeout => 1.seconds, :attempts => 2, :store => @no_store)
-      action = lambda{|*args| while true; end}
-      handler = Handler.create(action)
-      result = message.process(handler)
-      assert_equal RC::ExceptionsLimitReached, result
+  class HandlerTimeoutWitoutDedupStoreTest < HandlerTimeoutTest
+    def setup
+      @store = NoDeduplicationStore.new
+      @store.flushdb
     end
-
-
   end
 
   class MySQLFailoverTest < Minitest::Test

From ce4386f1dc03545a524ba825b543eaf791d43a62 Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Mon, 17 Mar 2025 15:35:47 +0100
Subject: [PATCH 03/10] Try it again

---
 test/beetle/message_test.rb | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/test/beetle/message_test.rb b/test/beetle/message_test.rb
index 9d5a4c39..7876706a 100644
--- a/test/beetle/message_test.rb
+++ b/test/beetle/message_test.rb
@@ -651,6 +651,13 @@ def setup
 
   end
 
+  class SeenMessageNoDeduplicationStoreTest < SeenMessageTest
+    def setup
+      @store = NoDeduplicationStore.new
+      @store.flushdb
+    end
+  end
+
   class ProcessingTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -719,6 +726,13 @@ def setup
 
   end
 
+  class ProcessingNoDeduplicationStoreTest < ProcessingTest
+    def setup
+      @store = NoDeduplicationStore.new
+      @store.flushdb
+    end
+  end
+
   class HandlerTimeoutTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new

From 7e8a51770dd6b13bbfa29c926ea01da727d5f040 Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Mon, 17 Mar 2025 16:46:08 +0100
Subject: [PATCH 04/10] WIP

---
 lib/beetle/no_deduplication_store.rb |  8 +++-
 test/beetle/message_test.rb          | 64 ++++++++++++++++++++++------
 2 files changed, 58 insertions(+), 14 deletions(-)

diff --git a/lib/beetle/no_deduplication_store.rb b/lib/beetle/no_deduplication_store.rb
index 866c4708..d225dd77 100644
--- a/lib/beetle/no_deduplication_store.rb
+++ b/lib/beetle/no_deduplication_store.rb
@@ -12,11 +12,11 @@ def redis
     end
 
     def set(msg_id, suffix, value)
-      @data[key(msg_id, suffix)] = value
+      @data[key(msg_id, suffix)] = value.to_s
     end
 
     def setnx(msg_id, suffix, value)
-      @data[key(msg_id, suffix)] ||= value
+      @data[key(msg_id, suffix)] ||= value.to_s
     end
 
     def setnx_completed!(_msg_id)
@@ -55,6 +55,10 @@ def del_keys(msg_id)
       end
     end
 
+    def exists(msg_id, suffix)
+      @data.key?(key(msg_id, suffix))
+    end
+
     def flushdb
       @data = {}
     end
diff --git a/test/beetle/message_test.rb b/test/beetle/message_test.rb
index 7876706a..70bfdb66 100644
--- a/test/beetle/message_test.rb
+++ b/test/beetle/message_test.rb
@@ -218,6 +218,14 @@ def setup
     end
   end
 
+  class KeyManagementNoDedupTest < KeyManagementTest
+    def setup
+      super
+      @store = NoDeduplicationStore.new(@config)
+      @store.flushdb
+    end
+  end
+
   class AckingTest < Minitest::Test
 
     def setup
@@ -277,7 +285,7 @@ def setup
       assert_nil @store.get(message.msg_id, :ack_count)
       message.process(@null_handler)
       assert message.redundant?
-      assert_equal "1", @store.get(message.msg_id, :ack_count)
+      assert_equal "1", @store.get(message.msg_id, :ack_count).to_s
     end
 
     test "acking a redundant message twice should remove the ack_count key" do
@@ -293,6 +301,14 @@ def setup
 
   end
 
+  class AckingNoDedupTest < AckingTest
+    def setup
+      super
+      @store = NoDeduplicationStore.new
+      @store.flushdb
+    end
+  end
+
   class FreshMessageTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -324,11 +340,18 @@ def setup
       message.expects(:completed!).in_sequence(s)
       header.expects(:ack).in_sequence(s)
       assert_equal RC::OK, message.__send__(:process_internal, proc)
-      assert_equal "1", @store.get(message.msg_id, :ack_count)
+      assert_equal "1", @store.get(message.msg_id, :ack_count).to_s
     end
 
   end
 
+  class FreshMessageNoDedupTest < FreshMessageTest
+    def setup
+      @store = NoDeduplicationStore.new
+      @store.flushdb
+    end
+  end
+
   class SimpleMessageTest < Minitest::Test
     def setup
       @config = Configuration.new
@@ -386,6 +409,16 @@ def setup
 
   end
 
+  class SimpleMessageNoDedupTest < SimpleMessageTest
+    def setup
+      @config = Configuration.new
+      @config.redis_server = Beetle.config.redis_server
+      @store = NoDeduplicationStore.new(@config)
+      @store.flushdb
+    end
+
+  end
+
   class HandlerCrashTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -405,9 +438,9 @@ def setup
       header.expects(:ack).never
       assert_equal RC::HandlerCrash, message.__send__(:process_internal, proc)
       assert !message.completed?
-      assert_equal "1", @store.get(message.msg_id, :exceptions)
-      assert_equal "0", @store.get(message.msg_id, :timeout)
-      assert_equal "52", @store.get(message.msg_id, :delay)
+      assert_equal "1", @store.get(message.msg_id, :exceptions).to_s
+      assert_equal "0", @store.get(message.msg_id, :timeout).to_s
+      assert_equal "52", @store.get(message.msg_id, :delay).to_s
     end
 
     test "a message should delete the mutex before resetting the timer if attempts and exception limits haven't been reached" do
@@ -494,6 +527,13 @@ class OtherException < StandardError; end
 
   end
 
+  class HandlerCrashNoDedupTest < HandlerCrashTest
+    def setup
+      @store = NoDeduplicationStore.new
+      @store.flushdb
+    end
+  end
+
   class SeenMessageTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -651,13 +691,13 @@ def setup
 
   end
 
-  class SeenMessageNoDeduplicationStoreTest < SeenMessageTest
-    def setup
-      @store = NoDeduplicationStore.new
-      @store.flushdb
-    end
-  end
-
+  # class SeenMessageNoDedupTest < SeenMessageTest
+  #   def setup
+  #     @store = NoDeduplicationStore.new
+  #     @store.flushdb
+  #   end
+  # end
+  #
   class ProcessingTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new

From edac4d59737d8331937791453ea57d25c2407403 Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Tue, 18 Mar 2025 09:35:35 +0100
Subject: [PATCH 05/10] Layout integration of single broker mode

---
 Gemfile                              |  1 +
 lib/beetle/base.rb                   |  4 ++
 lib/beetle/client.rb                 |  4 ++
 lib/beetle/message.rb                |  9 ++++
 lib/beetle/no_deduplication_store.rb | 68 ----------------------------
 lib/beetle/publisher.rb              | 43 +++++++++++++++---
 lib/beetle/single_broker_message.rb  | 19 ++++++++
 lib/beetle/subscriber.rb             |  6 ++-
 test/beetle/client_test.rb           | 17 +++++++
 test/beetle/message_test.rb          | 61 -------------------------
 10 files changed, 96 insertions(+), 136 deletions(-)
 delete mode 100644 lib/beetle/no_deduplication_store.rb
 create mode 100644 lib/beetle/single_broker_message.rb

diff --git a/Gemfile b/Gemfile
index 5eee0307..db939f38 100644
--- a/Gemfile
+++ b/Gemfile
@@ -2,5 +2,6 @@ source 'https://rubygems.org'
 
 gemspec
 
+gem "pry"
 gem "hiredis-client"
 # gem 'bunny', '=0.7.10', :path => "#{ENV['HOME']}/src/bunny"
diff --git a/lib/beetle/base.rb b/lib/beetle/base.rb
index ea5f49db..5d3cf774 100644
--- a/lib/beetle/base.rb
+++ b/lib/beetle/base.rb
@@ -18,6 +18,10 @@ def initialize(client, options = {}) #:nodoc:
 
     private
 
+    def single_broker_mode?
+      @client.single_broker_mode?
+    end
+
     def error(text)
       logger.error text
       raise Error.new(text)
diff --git a/lib/beetle/client.rb b/lib/beetle/client.rb
index 2c62b88a..85f61635 100644
--- a/lib/beetle/client.rb
+++ b/lib/beetle/client.rb
@@ -68,6 +68,10 @@ def initialize(config = Beetle.config)
       )
     end
 
+    def single_broker_mode?
+      servers.size == 1
+    end
+
     # register an exchange with the given _name_ and a set of _options_:
     # [<tt>:type</tt>]
     #   the type option will be overwritten and always be <tt>:topic</tt>, beetle does not allow fanout exchanges
diff --git a/lib/beetle/message.rb b/lib/beetle/message.rb
index aa047166..d1da3518 100644
--- a/lib/beetle/message.rb
+++ b/lib/beetle/message.rb
@@ -62,6 +62,14 @@ class Message
     # value returned by handler execution
     attr_reader :handler_result
 
+    def self.create(queue, header, body, opts = {})
+      new(queue, header, body, opts)
+    end
+
+    def self.single_broker(queue, header, body, opts = {})
+      SingleBrokerMessage.new(queue, header, body, opts)
+    end
+
     def initialize(queue, header, body, opts = {})
       @queue  = queue
       @header = header
@@ -101,6 +109,7 @@ def decode #:nodoc:
     def self.publishing_options(opts = {}) #:nodoc:
       flags = 0
       flags |= FLAG_REDUNDANT if opts[:redundant]
+
       expires_at = now + (opts[:ttl] || DEFAULT_TTL).to_i
       opts = opts.slice(*PUBLISHING_KEYS)
       opts[:message_id] = generate_uuid.to_s
diff --git a/lib/beetle/no_deduplication_store.rb b/lib/beetle/no_deduplication_store.rb
deleted file mode 100644
index d225dd77..00000000
--- a/lib/beetle/no_deduplication_store.rb
+++ /dev/null
@@ -1,68 +0,0 @@
-module Beetle
-  class NoDeduplicationStore < DeduplicationStore
-    include Logging
-
-    def initialize(config = Beetle.config)
-      super(config)
-      @data = {}
-    end
-
-    def redis
-      raise NotImplementedError
-    end
-
-    def set(msg_id, suffix, value)
-      @data[key(msg_id, suffix)] = value.to_s
-    end
-
-    def setnx(msg_id, suffix, value)
-      @data[key(msg_id, suffix)] ||= value.to_s
-    end
-
-    def setnx_completed!(_msg_id)
-      true
-    end
-
-    def mset(msg_id, values)
-      values.each do |k, v|
-        set(msg_id, k, v)
-      end
-    end
-
-    def msetnx(msg_id, values)
-      values.map { |k, _v| key(msg_id, k) }
-
-      return unless values.keys.intersection(@data.keys).empty?
-      mset(msg_id, values)
-    end
-
-    def incr(msg_id, suffix)
-      @data[key(msg_id, suffix)] ||= 0
-      @data[key(msg_id, suffix)] += 1
-    end
-
-    def get(msg_id, suffix)
-      @data[key(msg_id, suffix)]
-    end
-
-    def del(msg_id, suffix)
-      @data[key(msg_id, suffix)] = nil
-    end
-
-    def del_keys(msg_id)
-      keys(msg_id).each do |key|
-        @data[key] = nil
-      end
-    end
-
-    def exists(msg_id, suffix)
-      @data.key?(key(msg_id, suffix))
-    end
-
-    def flushdb
-      @data = {}
-    end
-
-    def with_failover; end
-  end
-end
diff --git a/lib/beetle/publisher.rb b/lib/beetle/publisher.rb
index db4dd3d4..9f882f31 100644
--- a/lib/beetle/publisher.rb
+++ b/lib/beetle/publisher.rb
@@ -39,16 +39,47 @@ def publish(message_name, data, opts={}) #:nodoc:
         opts = @client.messages[message_name].merge(opts.symbolize_keys)
         exchange_name = opts.delete(:exchange)
         opts.delete(:queue)
-        recycle_dead_servers unless @dead_servers.empty?
-        throttle!
-        if opts[:redundant]
-          publish_with_redundancy(exchange_name, message_name, data.to_s, opts)
+
+        if single_broker_mode?
+          throttle!
+          # TODO: add circuit breaker here to give a single server time to recover. `with_circuit_breaker { }`
+          publish_single_broker(exchange_name, message_name, data.to_s, opts) 
         else
-          publish_with_failover(exchange_name, message_name, data.to_s, opts)
+          recycle_dead_servers unless @dead_servers.empty?
+          throttle!
+          if opts[:redundant]
+            publish_with_redundancy(exchange_name, message_name, data.to_s, opts)
+          else
+            publish_with_failover(exchange_name, message_name, data.to_s, opts)
+          end
         end
       end
     end
 
+    def publish_single_broker(exchange_name, message_name, data, opts) #:nodoc:
+      published = 0
+      tries = 2
+      logger.debug "Beetle: sending #{message_name}"
+      opts = Message.publishing_options(opts.merge(redundant: false))
+
+      begin
+        select_next_server 
+        bind_queues_for_exchange(exchange_name)
+        logger.debug "Beetle: trying to send message #{message_name}: #{data} with option #{opts}"
+        exchange(exchange_name).publish(data, opts.dup)
+        logger.debug "Beetle: message sent!"
+        published = 1
+      rescue *bunny_exceptions => e
+        logger.warn("Beetle: publishing exception #{e} #{e.backtrace[0..4].join("\n")}")
+        stop!(e)
+        tries -= 1
+        retry if tries > 0
+        logger.error "Beetle: message could not be delivered: #{message_name}"
+        raise NoMessageSent.new
+      end
+      published
+    end
+
     def publish_with_failover(exchange_name, message_name, data, opts) #:nodoc:
       tries = @servers.size * 2
       logger.debug "Beetle: sending #{message_name}"
@@ -218,7 +249,7 @@ def select_next_server
       else
         set_current_server(@servers[((@servers.index(@server) || 0)+1) % @servers.size])
       end
-      logger.debug("Selected new server for publishing:#{server}.\n Dead servers are #{dead_servers.keys.any? ? dead_servers.keys.join(', ') : "none"}")
+      logger.debug("Selected new server for publishing:#{server}.\n Dead servers are #{dead_servers.keys.any? ? dead_servers.keys.join(', ') : 'none'}")
     end
 
     def create_exchange!(name, opts)
diff --git a/lib/beetle/single_broker_message.rb b/lib/beetle/single_broker_message.rb
new file mode 100644
index 00000000..204b5f66
--- /dev/null
+++ b/lib/beetle/single_broker_message.rb
@@ -0,0 +1,19 @@
+module Beetle
+  class SingleBrokerMessage < Message
+    def set_timeout!; end
+    def timed_out?(_t = nil) = false
+    def timed_out!; end
+
+    def redundant?
+      false
+    end
+
+    def aquire_mutex!
+      true
+    end
+
+    def delete_mutex!
+      true
+    end
+  end
+end
diff --git a/lib/beetle/subscriber.rb b/lib/beetle/subscriber.rb
index 46b8f26a..aaad7f19 100644
--- a/lib/beetle/subscriber.rb
+++ b/lib/beetle/subscriber.rb
@@ -177,7 +177,11 @@ def create_subscription_callback(queue_name, amqp_queue_name, handler, opts)
         end
         begin
           message_options = opts.merge(:server => server, :store => @client.deduplication_store)
-          m = Message.new(amqp_queue_name, header, data, message_options)
+          m = if single_broker_mode?
+                SingleBrokerMessage.create(amqp_queue_name, header, data, message_options)
+              else
+                Message.create(amqp_queue_name, header, data, message_options)
+              end
           processor = Handler.create(handler, opts)
           result = m.process(processor)
           if result.reject?
diff --git a/test/beetle/client_test.rb b/test/beetle/client_test.rb
index 4bd837f1..197fb5ad 100644
--- a/test/beetle/client_test.rb
+++ b/test/beetle/client_test.rb
@@ -476,4 +476,21 @@ def setup
     end
   end
 
+  class SingleBrokerModeTest < Minitest::Test
+    test "when only one server is configured we set single broker mode" do
+      config = Configuration.new
+      config.servers = "localhost:5672"
+      client = Client.new(config)
+
+      assert client.single_broker_mode?
+    end
+
+    test "when more than one server is configured we set single broker mode to false" do
+      config = Configuration.new
+      config.servers = "localhost:5672,localhost:5673"
+      client = Client.new(config)
+      refute client.single_broker_mode?
+    end
+  end
+
 end
diff --git a/test/beetle/message_test.rb b/test/beetle/message_test.rb
index 70bfdb66..0b667089 100644
--- a/test/beetle/message_test.rb
+++ b/test/beetle/message_test.rb
@@ -218,14 +218,6 @@ def setup
     end
   end
 
-  class KeyManagementNoDedupTest < KeyManagementTest
-    def setup
-      super
-      @store = NoDeduplicationStore.new(@config)
-      @store.flushdb
-    end
-  end
-
   class AckingTest < Minitest::Test
 
     def setup
@@ -301,14 +293,6 @@ def setup
 
   end
 
-  class AckingNoDedupTest < AckingTest
-    def setup
-      super
-      @store = NoDeduplicationStore.new
-      @store.flushdb
-    end
-  end
-
   class FreshMessageTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -345,13 +329,6 @@ def setup
 
   end
 
-  class FreshMessageNoDedupTest < FreshMessageTest
-    def setup
-      @store = NoDeduplicationStore.new
-      @store.flushdb
-    end
-  end
-
   class SimpleMessageTest < Minitest::Test
     def setup
       @config = Configuration.new
@@ -409,16 +386,6 @@ def setup
 
   end
 
-  class SimpleMessageNoDedupTest < SimpleMessageTest
-    def setup
-      @config = Configuration.new
-      @config.redis_server = Beetle.config.redis_server
-      @store = NoDeduplicationStore.new(@config)
-      @store.flushdb
-    end
-
-  end
-
   class HandlerCrashTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -527,13 +494,6 @@ class OtherException < StandardError; end
 
   end
 
-  class HandlerCrashNoDedupTest < HandlerCrashTest
-    def setup
-      @store = NoDeduplicationStore.new
-      @store.flushdb
-    end
-  end
-
   class SeenMessageTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -691,13 +651,6 @@ def setup
 
   end
 
-  # class SeenMessageNoDedupTest < SeenMessageTest
-  #   def setup
-  #     @store = NoDeduplicationStore.new
-  #     @store.flushdb
-  #   end
-  # end
-  #
   class ProcessingTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -766,13 +719,6 @@ def setup
 
   end
 
-  class ProcessingNoDeduplicationStoreTest < ProcessingTest
-    def setup
-      @store = NoDeduplicationStore.new
-      @store.flushdb
-    end
-  end
-
   class HandlerTimeoutTest < Minitest::Test
     def setup
       @store = DeduplicationStore.new
@@ -800,13 +746,6 @@ def setup
     end
   end
 
-  class HandlerTimeoutWitoutDedupStoreTest < HandlerTimeoutTest
-    def setup
-      @store = NoDeduplicationStore.new
-      @store.flushdb
-    end
-  end
-
   class MySQLFailoverTest < Minitest::Test
     require "active_record"
 

From 0ee63b135d875475cd50626ae7eb67a2595ce252 Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Tue, 18 Mar 2025 09:37:19 +0100
Subject: [PATCH 06/10] format

---
 test/beetle/client_test.rb | 1 +
 1 file changed, 1 insertion(+)

diff --git a/test/beetle/client_test.rb b/test/beetle/client_test.rb
index 197fb5ad..7b3e74f4 100644
--- a/test/beetle/client_test.rb
+++ b/test/beetle/client_test.rb
@@ -489,6 +489,7 @@ class SingleBrokerModeTest < Minitest::Test
       config = Configuration.new
       config.servers = "localhost:5672,localhost:5673"
       client = Client.new(config)
+
       refute client.single_broker_mode?
     end
   end

From 20e8c4033baeadbdc49d80ce5968170844545bd8 Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Tue, 18 Mar 2025 14:55:03 +0100
Subject: [PATCH 07/10] Add first change to SingleBrokerMessage

---
 lib/beetle/single_broker_message.rb | 58 +++++++++++++++++++++++++++++
 test/beetle/message_test.rb         | 40 +++++++++++++++++++-
 2 files changed, 97 insertions(+), 1 deletion(-)

diff --git a/lib/beetle/single_broker_message.rb b/lib/beetle/single_broker_message.rb
index 204b5f66..75b07b8d 100644
--- a/lib/beetle/single_broker_message.rb
+++ b/lib/beetle/single_broker_message.rb
@@ -1,5 +1,10 @@
 module Beetle
   class SingleBrokerMessage < Message
+    def ack!
+      logger.debug "Beetle: ack! for message #{msg_id}"
+      header.ack
+    end
+
     def set_timeout!; end
     def timed_out?(_t = nil) = false
     def timed_out!; end
@@ -15,5 +20,58 @@ def aquire_mutex!
     def delete_mutex!
       true
     end
+
+    private
+
+    def process_internal(handler)
+      if @exception
+        ack!
+        RC::DecodingError
+      elsif @pre_exception
+        ack!
+        RC::PreprocessingError
+      elsif expired?
+        logger.warn "Beetle: ignored expired message (#{msg_id})!"
+        ack!
+        RC::Ancient
+      elsif simple?
+        ack!
+        if @store.setnx_completed!(msg_id)
+          run_handler(handler) == RC::HandlerCrash ? RC::AttemptsLimitReached : RC::OK
+        else
+          RC::OK
+        end
+      elsif !key_exists?
+        run_handler!(handler)
+      else
+        status, delay, timeout, attempts, exceptions = fetch_status_delay_timeout_attempts_exceptions
+        if status == "completed"
+          ack!
+          RC::OK
+        elsif delay && delayed?(delay)
+          logger.warn "Beetle: ignored delayed message (#{msg_id})!"
+          RC::Delayed
+        elsif !(timeout && timed_out?(timeout))
+          RC::HandlerNotYetTimedOut
+        elsif attempts && attempts_limit_reached?(attempts)
+          completed!
+          ack!
+          logger.warn "Beetle: reached the handler execution attempts limit: #{attempts_limit} on #{msg_id}"
+          RC::AttemptsLimitReached
+        elsif exceptions && exceptions_limit_reached?(exceptions)
+          completed!
+          ack!
+          logger.warn "Beetle: reached the handler exceptions limit: #{exceptions_limit} on #{msg_id}"
+          RC::ExceptionsLimitReached
+        else
+          set_timeout!
+          if aquire_mutex!
+            run_handler!(handler)
+          else
+            RC::MutexLocked
+          end
+        end
+      end
+    end
   end
 end
diff --git a/test/beetle/message_test.rb b/test/beetle/message_test.rb
index 0b667089..5cb79ecd 100644
--- a/test/beetle/message_test.rb
+++ b/test/beetle/message_test.rb
@@ -219,7 +219,6 @@ def setup
   end
 
   class AckingTest < Minitest::Test
-
     def setup
       @store = DeduplicationStore.new
       @store.flushdb
@@ -237,6 +236,17 @@ def setup
       assert_equal :no, processed
     end
 
+    test "[SingleBroker] an expired message should be acked without calling the handler" do
+      header = header_with_params(:ttl => -1)
+      header.expects(:ack)
+      message = SingleBrokerMessage.new("somequeue", header, 'foo', :store => @store)
+      assert message.expired?
+
+      processed = :no
+      message.process(Handler.create(lambda {|*args| processed = true}))
+      assert_equal :no, processed
+    end
+
     test "a delayed message should not be acked and the handler should not be called" do
       header = header_with_params()
       header.expects(:ack).never
@@ -250,6 +260,20 @@ def setup
       assert_equal :no, processed
     end
 
+    test "[SingleBroker] a delayed message should not be acked and the handler should not be called" do
+      header = header_with_params()
+      header.expects(:ack).never
+      message = SingleBrokerMessage.new("somequeue", header, 'foo', :attempts => 2, :store => @store)
+      message.set_delay!
+      assert !message.key_exists?
+      assert message.delayed?
+
+      processed = :no
+      message.process(Handler.create(lambda {|*args| processed = true}))
+      assert_equal :no, processed
+    end
+
+
     test "acking a non redundant message should remove the ack_count key" do
       header = header_with_params({})
       header.expects(:ack)
@@ -312,6 +336,20 @@ def setup
       assert_equal RC::OK, message.process(handler)
     end
 
+    test "[SingleBroker] processing a fresh message sucessfully should first run the handler and then ack it" do
+      header = header_with_params({})
+      message = SingleBrokerMessage.new("somequeue", header, 'foo', :attempts => 2, :store => @store)
+      assert !message.attempts_limit_reached?
+
+      handler = mock("handler")
+      s = sequence("s")
+      handler.expects(:pre_process).with(message).in_sequence(s)
+      handler.expects(:call).in_sequence(s)
+      header.expects(:ack).in_sequence(s)
+      assert_equal RC::OK, message.process(handler)
+    end
+
+
     test "after processing a redundant fresh message successfully the ack count should be 1 and the status should be completed" do
       header = header_with_params({:redundant => true})
       message = Message.new("somequeue", header, 'foo', :timeout => 10.seconds, :store => @store)

From 77666540c357761f0f2055e4bd584947a3c02109 Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Tue, 18 Mar 2025 16:47:13 +0100
Subject: [PATCH 08/10] WIP

---
 lib/beetle/single_broker_message.rb | 106 +++++++++++++++++-----------
 test/beetle/bunny_behavior_test.rb  |  59 +++++++---------
 test/beetle/message_test.rb         |  65 ++++++++++++++---
 3 files changed, 146 insertions(+), 84 deletions(-)

diff --git a/lib/beetle/single_broker_message.rb b/lib/beetle/single_broker_message.rb
index 75b07b8d..8a476e5c 100644
--- a/lib/beetle/single_broker_message.rb
+++ b/lib/beetle/single_broker_message.rb
@@ -1,28 +1,76 @@
 module Beetle
   class SingleBrokerMessage < Message
+    def initialize(*args, **kwargs)
+      super(*args, **kwargs)
+      @store = nil
+    end
+
     def ack!
       logger.debug "Beetle: ack! for message #{msg_id}"
       header.ack
     end
 
-    def set_timeout!; end
-    def timed_out?(_t = nil) = false
-    def timed_out!; end
+    def set_delay!
+      log_not_supported("delay between retries")
+    end
+
+    def delayed?
+      log_not_supported("delay between retries")
+      false
+    end
 
     def redundant?
       false
     end
 
-    def aquire_mutex!
-      true
+    def increment_execution_attempts!; end
+
+    def attempts_limit_reached?(_attempts = nil)
+      # TODO: implement
+      false
     end
 
-    def delete_mutex!
-      true
+    def exceptions_limit_reached?
+      # TODO: implement
+      false
     end
 
     private
 
+    def log_not_supported(what)
+      logger.warn "Beetle: Feature not supported in single broker mode => #{what}"
+    end
+
+    def run_handler!(handler)
+      case result = run_handler(handler)
+      when RC::OK
+        ack!
+        result
+      else
+        handler_failed!(result)
+      end
+    end
+
+    def handler_failed!(result)
+      if attempts_limit_reached?
+        ack!
+        logger.debug "Beetle: reached the handler execution attempts limit: #{attempts_limit} on #{msg_id}"
+        RC::AttemptsLimitReached
+      elsif exceptions_limit_reached?
+        ack!
+        logger.debug "Beetle: reached the handler exceptions limit: #{exceptions_limit} on #{msg_id}"
+        RC::ExceptionsLimitReached
+      elsif !exception_accepted?
+        ack!
+        logger.debug "Beetle: `#{@exception.class.name}` not accepted: `retry_on`=[#{retry_on.join(',')}] on #{msg_id}"
+        RC::ExceptionNotAccepted
+      else
+        result
+      end
+    end
+
+    # Open questions:
+    # - do we need to support timeouts that span executions?
     def process_internal(handler)
       if @exception
         ack!
@@ -36,41 +84,17 @@ def process_internal(handler)
         RC::Ancient
       elsif simple?
         ack!
-        if @store.setnx_completed!(msg_id)
-          run_handler(handler) == RC::HandlerCrash ? RC::AttemptsLimitReached : RC::OK
-        else
-          RC::OK
-        end
-      elsif !key_exists?
-        run_handler!(handler)
+        run_handler(handler) == RC::HandlerCrash ? RC::AttemptsLimitReached : RC::OK
+      elsif attempts_limit_reached?
+        ack!
+        logger.warn "Beetle: reached the handler execution attempts limit: #{attempts_limit} on #{msg_id}"
+        RC::AttemptsLimitReached
+      elsif exceptions_limit_reached?
+        ack!
+        logger.warn "Beetle: reached the handler exceptions limit: #{exceptions_limit} on #{msg_id}"
+        RC::ExceptionsLimitReached
       else
-        status, delay, timeout, attempts, exceptions = fetch_status_delay_timeout_attempts_exceptions
-        if status == "completed"
-          ack!
-          RC::OK
-        elsif delay && delayed?(delay)
-          logger.warn "Beetle: ignored delayed message (#{msg_id})!"
-          RC::Delayed
-        elsif !(timeout && timed_out?(timeout))
-          RC::HandlerNotYetTimedOut
-        elsif attempts && attempts_limit_reached?(attempts)
-          completed!
-          ack!
-          logger.warn "Beetle: reached the handler execution attempts limit: #{attempts_limit} on #{msg_id}"
-          RC::AttemptsLimitReached
-        elsif exceptions && exceptions_limit_reached?(exceptions)
-          completed!
-          ack!
-          logger.warn "Beetle: reached the handler exceptions limit: #{exceptions_limit} on #{msg_id}"
-          RC::ExceptionsLimitReached
-        else
-          set_timeout!
-          if aquire_mutex!
-            run_handler!(handler)
-          else
-            RC::MutexLocked
-          end
-        end
+        run_handler!(handler)
       end
     end
   end
diff --git a/test/beetle/bunny_behavior_test.rb b/test/beetle/bunny_behavior_test.rb
index 6149e977..d3ac7e7c 100644
--- a/test/beetle/bunny_behavior_test.rb
+++ b/test/beetle/bunny_behavior_test.rb
@@ -26,8 +26,6 @@ class BunnyBehaviorTest < Minitest::Test
     assert_equal({"bar" => "baz"}, headers["table"])
   end
 
-
-  
   test "publishing redundantly does not leave the garbage in dedup store" do
     Beetle.config.servers = "localhost:5672,localhost:5673"
     client = Beetle::Client.new
@@ -38,7 +36,7 @@ class BunnyBehaviorTest < Minitest::Test
     # empty the dedup store
     client.deduplication_store.flushdb
 
-    handler = TestHandler.new(stop_listening_after_n_post_processes = 2, client = client)
+    handler = TestHandler.new(2, client = client)
     client.register_handler(:test_garbage, handler)
     published = client.publish(:test_garbage, 'bam', :redundant =>true)
     listen(client)
@@ -49,9 +47,9 @@ class BunnyBehaviorTest < Minitest::Test
     message = messages_processed.first
     assert_equal 2, published
     assert_equal "bam", message.data
-    Beetle::DeduplicationStore::KEY_SUFFIXES.each{|suffix| 
-        assert_equal false, client.deduplication_store.exists(message.msg_id, suffix)
-    }
+    Beetle::DeduplicationStore::KEY_SUFFIXES.each do |suffix|
+      assert_equal false, client.deduplication_store.exists(message.msg_id, suffix)
+    end
   end
 
   test "process redundant message once" do
@@ -64,7 +62,7 @@ class BunnyBehaviorTest < Minitest::Test
     # empty the dedup store
     client.deduplication_store.flushdb
 
-    handler = TestHandler.new(stop_listening_after_n_post_processes = 2, client = client)
+    handler = TestHandler.new(2, client = client)
     client.register_handler(:test_processing, handler)
     published = client.publish(:test_processing, 'bam', :redundant =>true)
     listen(client)
@@ -98,16 +96,20 @@ class BunnyBehaviorTest < Minitest::Test
 
     assert_equal 1, published
     assert_equal "bam", message.data
-    Beetle::DeduplicationStore::KEY_SUFFIXES.map{|suffix| 
-        assert_equal false, client.deduplication_store.exists(message.msg_id, suffix)
-    }
+    Beetle::DeduplicationStore::KEY_SUFFIXES.map do |suffix|
+      assert_equal false, client.deduplication_store.exists(message.msg_id, suffix)
+    end
   end
 
+  # FIXME: that's not really testing that publisher confirms work
   test "publishing with confirms works as expected" do
-    Beetle.config.servers = "localhost:5672"
-    client = Beetle::Client.new
+    config = Beetle::Configuration.new
+    config.servers = "localhost:5672"
+    config.publisher_confirms = true
+    client = Beetle::Client.new(config)
+
     client.register_queue(:test_publisher_confirms)
-    client.register_message(:test_publisher_confirms, :publisher_confirms => true)
+    client.register_message(:test_publisher_confirms)
     # purge the test queue
     client.purge(:test_publisher_confirms)
 
@@ -119,26 +121,21 @@ class BunnyBehaviorTest < Minitest::Test
     client.stop_publishing
 
     assert_equal 1, published
-    assert_equal "bam", message.data 
-
+    assert_equal "bam", message.data
   end
 
-
-  def listen(client , timeout = 1) 
-    Timeout.timeout(timeout) do 
-      client.listen 
+  def listen(client, timeout = 1)
+    Timeout.timeout(timeout) do
+      client.listen
     end
-  rescue Timeout::Error 
-       puts "Client listen timed out after #{timeout} seconds"
-       nil
+  rescue Timeout::Error
+    puts "Client listen timed out after #{timeout} seconds"
+    nil
   end
 
-
   class TestHandler < Beetle::Handler
 
-    attr_reader :messages_processed
-    attr_reader :pre_process_invocations
-    attr_reader :post_process_invocations
+    attr_reader :messages_processed, :pre_process_invocations, :post_process_invocations
 
     def initialize(stop_listening_after_n_post_processes, client)
       super()
@@ -149,22 +146,20 @@ def initialize(stop_listening_after_n_post_processes, client)
       @messages_processed = []
     end
 
-    def pre_process(message)
+    def pre_process(_message)
       @pre_process_invocations += 1
     end
-    
+
     def process
       @messages_processed << message
     end
 
     def post_process
       @post_process_invocations += 1
-      if @post_process_invocations >= @stop_listening_after_n_post_processes
-          @client.stop_listening
-      end
+      return unless @post_process_invocations >= @stop_listening_after_n_post_processes
+      @client.stop_listening
     end
 
   end
 
-
 end
diff --git a/test/beetle/message_test.rb b/test/beetle/message_test.rb
index 5cb79ecd..0d44b3f7 100644
--- a/test/beetle/message_test.rb
+++ b/test/beetle/message_test.rb
@@ -261,19 +261,9 @@ def setup
     end
 
     test "[SingleBroker] a delayed message should not be acked and the handler should not be called" do
-      header = header_with_params()
-      header.expects(:ack).never
-      message = SingleBrokerMessage.new("somequeue", header, 'foo', :attempts => 2, :store => @store)
-      message.set_delay!
-      assert !message.key_exists?
-      assert message.delayed?
-
-      processed = :no
-      message.process(Handler.create(lambda {|*args| processed = true}))
-      assert_equal :no, processed
+      assert true, "Not supported in single broker mode"
     end
 
-
     test "acking a non redundant message should remove the ack_count key" do
       header = header_with_params({})
       header.expects(:ack)
@@ -284,6 +274,10 @@ def setup
       assert !@store.exists(message.msg_id, :ack_count)
     end
 
+    test "[SingleBroker] acking a non redundant message should remove the ack_count key" do
+      assert true, "Not supported in single broker mode. No redundant messages."
+    end
+
     test "a redundant message should be acked after calling the handler" do
       header = header_with_params({:redundant => true})
       message = Message.new("somequeue", header, 'foo', :store => @store)
@@ -293,6 +287,10 @@ def setup
       message.process(@null_handler)
     end
 
+    test "[SingleBroker] a redundant message should be acked after calling the handler" do
+      assert true, "Not supported in single broker mode. No redundant messages."
+    end
+
     test "acking a redundant message should increment the ack_count key" do
       header = header_with_params({:redundant => true})
       header.expects(:ack)
@@ -304,6 +302,11 @@ def setup
       assert_equal "1", @store.get(message.msg_id, :ack_count).to_s
     end
 
+
+    test "[SingleBroker] acking a redundant message should increment the ack_count key" do
+      assert true, "Not supported in single broker mode. No redundant messages."
+    end
+
     test "acking a redundant message twice should remove the ack_count key" do
       header = header_with_params({:redundant => true})
       header.expects(:ack).twice
@@ -315,6 +318,10 @@ def setup
       assert !@store.exists(message.msg_id, :ack_count)
     end
 
+
+    test "[SingleBroker] acking a redundant message twice should remove the ack_count key" do
+      assert true, "Not supported in single broker mode. No redundant messages."
+    end
   end
 
   class FreshMessageTest < Minitest::Test
@@ -365,8 +372,14 @@ def setup
       assert_equal "1", @store.get(message.msg_id, :ack_count).to_s
     end
 
+
+    test "[SingleBroker] after processing a redundant fresh message successfully the ack count should be 1 and the status should be completed" do
+
+      assert true, "Not supported in single broker mode. No redundant messages."
+    end
   end
 
+
   class SimpleMessageTest < Minitest::Test
     def setup
       @config = Configuration.new
@@ -387,6 +400,20 @@ def setup
       assert_equal RC::OK, message.process(handler)
     end
 
+   test "[SingleBroker] when processing a simple message, ack should follow calling the handler" do
+      header = header_with_params({})
+      message = SingleBrokerMessage.new("somequeue", header, 'foo', :attempts => 1, :store => nil)
+
+      handler = mock("handler")
+      s = sequence("s")
+      handler.expects(:pre_process).with(message).in_sequence(s)
+      header.expects(:ack).in_sequence(s)
+      handler.expects(:call).in_sequence(s)
+      assert_equal RC::OK, message.process(handler)
+    end
+
+
+
     test "when processing a simple message, RC::AttemptsLimitReached should be returned if the handler crashes" do
       header = header_with_params({})
       message = Message.new("somequeue", header, 'foo', :attempts => 1, :store => @store)
@@ -402,6 +429,22 @@ def setup
       assert_equal RC::AttemptsLimitReached, message.process(handler)
     end
 
+    test "[SingleBroker] when processing a simple message, RC::AttemptsLimitReached should be returned if the handler crashes" do
+      header = header_with_params({})
+      message = SingleBrokerMessage.new("somequeue", header, 'foo', :attempts => 1, :store => nil)
+
+      handler = mock("handler")
+      s = sequence("s")
+      handler.expects(:pre_process).with(message).in_sequence(s)
+      header.expects(:ack).in_sequence(s)
+      e = Exception.new("ohoh")
+      handler.expects(:call).in_sequence(s).raises(e)
+      handler.expects(:process_exception).with(e).in_sequence(s)
+      handler.expects(:process_failure).with(RC::AttemptsLimitReached).in_sequence(s)
+      assert_equal RC::AttemptsLimitReached, message.process(handler)
+    end
+
+
     test "when processing a simple message, the handler should be executed only once if status keys are used" do
       @config.redis_status_key_expiry_interval = 1.minute
       header = header_with_params({})

From 4cb1f97c7efb5e2c4dcc56e7acb78546d125904f Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Tue, 18 Mar 2025 16:47:20 +0100
Subject: [PATCH 09/10] WIP

---
 .../single_broker_mode_behavior_test.rb       | 81 +++++++++++++++++++
 1 file changed, 81 insertions(+)
 create mode 100644 test/beetle/single_broker_mode_behavior_test.rb

diff --git a/test/beetle/single_broker_mode_behavior_test.rb b/test/beetle/single_broker_mode_behavior_test.rb
new file mode 100644
index 00000000..2f00ef44
--- /dev/null
+++ b/test/beetle/single_broker_mode_behavior_test.rb
@@ -0,0 +1,81 @@
+require 'timeout'
+require File.expand_path(File.dirname(__FILE__) + '/../test_helper')
+
+class SingleBrokerModeBehaviorTest < Minitest::Test
+  attr_reader :client
+
+  def setup
+    Beetle.config.servers = "localhost:5672"
+    @client = Beetle::Client.new
+    client.register_queue(:test_single_broker)
+    client.register_message(:test_single_broker)
+    client.purge(:test_single_broker)
+  end
+
+  test "handles retries correctly" do
+    assert true
+  end
+
+  test "handles attempts correctly" do
+    assert true
+  end
+
+  test "handles timeouts correctly" do
+    message = nil
+
+    client.register_handler(:test_single_broker, timeout: 1) do |msg|
+      sleep 1.5
+      message = msg
+      client.stop_listening
+    end
+    published = client.publish(:test_single_broker, 'single-broker')
+
+    listen(client, 2)
+    client.stop_publishing
+
+    assert_equal 1, published
+    refute message
+  end
+
+  test "ignores expired messages" do
+    message = nil
+
+    client.register_handler(:test_single_broker) do |msg|
+      message = msg
+      client.stop_listening
+    end
+    published = client.publish(:test_single_broker, 'single-broker', ttl: -3600) # expired 1 hour ago
+
+    listen(client)
+    client.stop_publishing
+
+    assert_equal 1, published
+    refute message
+  end
+
+  test "processes fresh message" do
+    message = nil
+
+    client.register_handler(:test_single_broker) do |msg|
+      message = msg
+      client.stop_listening
+    end
+    published = client.publish(:test_single_broker, 'single-broker', ttl: 30)
+
+    listen(client)
+    client.stop_publishing
+
+    assert_equal 1, published
+    assert_equal "single-broker", message.data
+  end
+
+  def listen(client, timeout = 1)
+    Timeout.timeout(timeout) do
+      client.listen
+    end
+  rescue Timeout::Error
+    puts "Client listen timed out after #{timeout} seconds"
+    nil
+  end
+
+end

From 78a83913ac8c0bd7f1b289fcae734b3a78ba4148 Mon Sep 17 00:00:00 2001
From: David Krentzlin <david.krentzlin@new-work.se>
Date: Tue, 18 Mar 2025 16:54:12 +0100
Subject: [PATCH 10/10] remove single broker publishing path for now

we'll revisit this at a later stage in this change
---
 lib/beetle/publisher.rb | 40 +++++-----------------------------------
 1 file changed, 5 insertions(+), 35 deletions(-)

diff --git a/lib/beetle/publisher.rb b/lib/beetle/publisher.rb
index 1f016508..6746173e 100644
--- a/lib/beetle/publisher.rb
+++ b/lib/beetle/publisher.rb
@@ -44,46 +44,16 @@ def publish(message_name, data, opts={}) #:nodoc:
         exchange_name = opts.delete(:exchange)
         opts.delete(:queue)
 
-        if single_broker_mode?
-          throttle!
-          # TODO: add circuit breaker here to give a single server time to recover. `with_circuit_breaker { }`
-          publish_single_broker(exchange_name, message_name, data.to_s, opts) 
+        recycle_dead_servers unless @dead_servers.empty?
+        throttle!
+        if opts[:redundant]
+          publish_with_redundancy(exchange_name, message_name, data.to_s, opts)
         else
-          recycle_dead_servers unless @dead_servers.empty?
-          throttle!
-          if opts[:redundant]
-            publish_with_redundancy(exchange_name, message_name, data.to_s, opts)
-          else
-            publish_with_failover(exchange_name, message_name, data.to_s, opts)
-          end
+          publish_with_failover(exchange_name, message_name, data.to_s, opts)
         end
       end
     end
 
-    def publish_single_broker(exchange_name, message_name, data, opts) #:nodoc:
-      published = 0
-      tries = 2
-      logger.debug "Beetle: sending #{message_name}"
-      opts = Message.publishing_options(opts.merge(redundant: false))
-
-      begin
-        select_next_server 
-        bind_queues_for_exchange(exchange_name)
-        logger.debug "Beetle: trying to send message #{message_name}: #{data} with option #{opts}"
-        exchange(exchange_name).publish(data, opts.dup)
-        logger.debug "Beetle: message sent!"
-        published = 1
-      rescue *bunny_exceptions => e
-        logger.warn("Beetle: publishing exception #{e} #{e.backtrace[0..4].join("\n")}")
-        stop!(e)
-        tries -= 1
-        retry if tries > 0
-        logger.error "Beetle: message could not be delivered: #{message_name}"
-        raise NoMessageSent.new
-      end
-      published
-    end
-
     def publish_with_failover(exchange_name, message_name, data, opts) #:nodoc:
       tries = @servers.size * 2
       logger.debug "Beetle: sending #{message_name}"