Add publisher confirms to support quorum queues.

.github/workflows/downstream.yml

@@ -2,8 +2,6 @@ name: Downstream
 
 on:
   workflow_dispatch:
-    branches:
-      - trunk
   push:
     branches:
       - trunk

.github/workflows/tests.yml

@@ -9,14 +9,14 @@ jobs:
     - uses: actions/checkout@v1
     - uses: ruby/setup-ruby@v1
       with:
-        ruby-version: 2.6.3
+        ruby-version: 2.6.9
     - name: Cache Gems
-      uses: actions/cache@v1
+      uses: actions/cache@v4
       with:
         path: vendor/bundle
-        key: ${{ runner.os }}-acapi-2.6.3-${{ hashFiles('**/Gemfile.lock') }}
+        key: ${{ runner.os }}-acapi-2.6.9-${{ hashFiles('**/Gemfile.lock') }}
         restore-keys: |
-          ${{ runner.os }}-acapi-2.6.3-${{ hashFiles('**/Gemfile.lock') }}
+          ${{ runner.os }}-acapi-2.6.9-${{ hashFiles('**/Gemfile.lock') }}
     - name: Install Gems
       run: |
         sudo gem install bundler -v '1.17.3'

.ruby-version

@@ -1 +1 @@
-2.6.6
+2.6.9

Gemfile.lock

@@ -55,7 +55,7 @@ GEM
       timers (~> 4.0.0)
     coderay (1.1.0)
     concurrent-ruby (1.1.6)
-    crass (1.0.4)
+    crass (1.0.6)
     diff-lcs (1.3)
     erubis (2.7.0)
     ffi (1.10.0)
@@ -85,20 +85,21 @@ GEM
       celluloid (~> 0.16.0)
       rb-fsevent (>= 0.9.3)
       rb-inotify (>= 0.9)
-    loofah (2.2.2)
+    loofah (2.25.0)
       crass (~> 1.0.2)
-      nokogiri (>= 1.5.9)
+      nokogiri (>= 1.12.0)
     lumberjack (1.0.9)
     mail (2.6.3)
       mime-types (>= 1.16, < 3)
     method_source (0.8.2)
     mime-types (2.99.3)
-    mini_portile2 (2.4.0)
+    mini_portile2 (2.8.9)
     minitest (5.14.0)
     multi_json (1.13.1)
     nenv (0.2.0)
-    nokogiri (1.9.1)
-      mini_portile2 (~> 2.4.0)
+    nokogiri (1.13.10)
+      mini_portile2 (~> 2.8.0)
+      racc (~> 1.4)
     notiffany (0.0.6)
       nenv (~> 0.1)
       shellany (~> 0.0)
@@ -111,6 +112,7 @@ GEM
     pry-remote (0.1.8)
       pry (~> 0.9)
       slop (~> 3.0)
+    racc (1.8.1)
     rack (1.6.9)
     rack-test (0.6.3)
       rack (>= 1.0)

lib/acapi.rb

@@ -2,6 +2,8 @@
 require "active_support"
 
 require "acapi/config"
+require "acapi/errors"
+
 require "acapi/notifiers"
 require "acapi/publisher"
 require "acapi/subscriber"

lib/acapi/amqp/client.rb

@@ -12,6 +12,9 @@ def initialize(chan, q)
         @argument_errors = []
         @bad_argument_queue = "acapi.error.middleware.service.bad_arguments"
         @processing_failed_queue = "acapi.error.middleware.service.processing_failed"
+        @republish_channel = @channel.connection.create_channel
+        @republish_channel.confirm_select
+        @republish_queue = @republish_channel.queue(@queue.name, @queue.options)
         @exit_after_work = false
       end
 
@@ -102,7 +105,8 @@ def subscribe(opts = {})
                 publish_processing_failed(delivery_info, properties, payload, e)
               else
                 new_properties = redelivery_properties(existing_retry_count, delivery_info, properties)
-                queue.publish(payload, new_properties)
+                @republish_queue.publish(payload, new_properties)
+                @republish_channel.wait_for_confirms || raise(Acapi::Errors::PublishConfirmationFailedError, "message republication could not be confirmed")
                 channel.acknowledge(delivery_info.delivery_tag, false)
               end
             rescue => e

lib/acapi/amqp/messaging_exchange_topology.rb

@@ -4,14 +4,14 @@ module Acapi
   module Amqp
     class MessagingExchangeTopology
 
-      def self.ensure_topology_exists(connection_string)
-        topology = new(connection_string)
+      def self.ensure_topology_exists(connection_settings)
+        topology = new(connection_settings)
         topology.setup
         topology.close
       end
 
-      def initialize(connection_string)
-        @connection = Bunny.new(connection_string, :heartbeat => 15)
+      def initialize(connection_settings)
+        @connection = Bunny.new(connection_settings)
         @connection.start
         @channel = @connection.create_channel
       end

lib/acapi/amqp/requestor.rb

@@ -11,10 +11,13 @@ def initialize(conn)
       def request(properties, payload, timeout = 15)
         delivery_info, r_props, r_payload = [nil, nil, nil]
         channel = @connection.create_channel
+        p_channel = @connection.create_channel
         temp_queue = channel.queue("", :exclusive => true)
         channel.prefetch(1)
-        request_exchange = channel.fanout(Rails.application.config.acapi.remote_request_exchange, :durable => true)
+        p_channel.confirm_select
+        request_exchange = p_channel.fanout(Rails.application.config.acapi.remote_request_exchange, :durable => true)
         request_exchange.publish(payload, properties.dup.merge({ :reply_to => temp_queue.name, :persistent => true }))
+        p_channel.wait_for_confirms || raise(Acapi::Errors::PublishConfirmationFailedError, "message publication could not be confirmed")
         delivery_info, r_props, r_payload = [nil, nil, nil]
         begin
           Timeout::timeout(timeout) do
@@ -26,6 +29,7 @@ def request(properties, payload, timeout = 15)
           end
         ensure
           temp_queue.delete
+          p_channel.close
           channel.close
         end
         [delivery_info, r_props, r_payload]

lib/acapi/amqp/responder.rb

@@ -3,8 +3,10 @@ module Amqp
     module Responder
       def with_response_exchange(connection)
         channel = connection.create_channel
+        channel.confirm_select
         publish_exchange = channel.default_exchange
         yield publish_exchange
+        channel.wait_for_confirms || raise(Acapi::Errors::PublishConfirmationFailedError, "message publication could not be confirmed")
         channel.close
       end
     end

lib/acapi/amqp_event_worker.rb

@@ -68,11 +68,11 @@ def self.run
       pid_file_location = File.join(File.expand_path(Rails.root), "pids", "sneakers.pid")
       worker_classes = Rails.application.config.acapi.sneakers_worker_classes
       ensure_messaging_exchanges
+      connection = Bunny.new(Rails.application.config.acapi.to_connection_settings)
       Sneakers.configure(
         :workers => worker_classes.length,
-        :amqp => Rails.application.config.acapi.remote_broker_uri,
+        :connection => connection,
         :start_worker_delay => 0.2,
-        :heartbeat => 5,
         :log => STDOUT,
         :pid_path => pid_file_location,
         :handler => Sneakers::Handlers::Maxretry,
@@ -88,7 +88,7 @@ def self.run
     end
 
     def self.ensure_messaging_exchanges
-      ::Acapi::Amqp::MessagingExchangeTopology.ensure_topology_exists(Rails.application.config.acapi.remote_broker_uri)
+      ::Acapi::Amqp::MessagingExchangeTopology.ensure_topology_exists(Rails.application.config.acapi.to_connection_settings)
     end
   end
 end

lib/acapi/errors.rb

@@ -0,0 +1,7 @@
+require "acapi/errors/publish_confirmation_failed_error"
+require "acapi/errors/remote_connection_unspecified_error"
+
+module Acapi
+  module Errors
+  end
+end

lib/acapi/errors/publish_confirmation_failed_error.rb

@@ -0,0 +1,5 @@
+module Acapi
+  module Errors
+    class PublishConfirmationFailedError < StandardError; end
+  end
+end

lib/acapi/errors/remote_connection_unspecified_error.rb

@@ -0,0 +1,5 @@
+module Acapi
+  module Errors
+    class RemoteConnectionUnspecifiedError < StandardError; end
+  end
+end

lib/acapi/local_amqp_publisher.rb

@@ -61,18 +61,25 @@ def log(name, started, finished, unique_id, data = {})
       end
       msg = Acapi::Amqp::OutMessage.new(@app_id, name, finished, finished, unique_id, data)
       @exchange.publish(*msg.to_message_properties)
+      @p_channel.wait_for_confirms || raise(Acapi::Errors::PublishConfirmationFailedError, "message publication could not be confirmed")
     end
 
     def open_connection_if_needed
       return if @connection.present? && @connection.connected?
-      @connection = Bunny.new
+      @connection = Bunny.new(connection_url)
       @connection.start
       @channel = @connection.create_channel
       @queue = @channel.queue(QUEUE_NAME, {:durable => true})
-      @exchange = @channel.fanout(EXCHANGE_NAME, {:durable => true})
+      @p_channel = @connection.create_channel
+      @p_channel.confirm_select
+      @exchange = @p_channel.fanout(EXCHANGE_NAME, {:durable => true})
       @queue.bind(@exchange, {})
     end
 
+    def connection_url
+      Rails.application.config.acapi.to_connection_settings
+    end
+
     def reconnect!
       disconnect!
     end
@@ -83,6 +90,10 @@ def disconnect!
           @connection.close
         rescue Timeout::Error
         end
+        @queue = nil
+        @channel = nil
+        @p_channel = nil
+        @exchange = nil
         @connection = nil
       end
     end

lib/acapi/publishers/upstream_event_publisher.rb

@@ -18,10 +18,9 @@ def run
         if @after_fork
           @after_fork.call
         end
-        bunny_url = Rails.application.config.acapi.remote_broker_uri
         event_q_name = Rails.application.config.acapi.remote_event_queue
         app_id = Rails.application.config.acapi.app_id
-        conn = Bunny.new(bunny_url, :heartbeat => 15)
+        conn = Bunny.new(Rails.application.config.acapi.to_connection_settings)
         conn.start
         chan = conn.create_channel
         chan.prefetch(1)

lib/acapi/railties/amqp_configuration_options.rb

@@ -1,10 +1,71 @@
+require 'uri'
+
 module Acapi
+  class ClusterSettings
+    attr_accessor :hosts
+    attr_accessor :port
+    attr_accessor :username
+    attr_accessor :password
+
+    def to_connection_settings
+      {
+        :hosts => @hosts,
+        :port => @port || 5672,
+        :username => @username || "guest",
+        :password => @password || "guest",
+        :heartbeat => 10
+      }
+    end
+  end
+
   class ConfigurationSettings
     attr_accessor :remote_broker_uri
     attr_accessor :remote_event_queue
     attr_accessor :remote_request_exchange
     attr_accessor :hbx_id
     attr_accessor :environment_name
+
+    def clear!
+      @remote_broker_uri = nil
+      @remote_event_queue = nil
+      @remote_request_exchange = nil
+      @hbx_id = nil
+      @environment_name = nil
+      @cluster = nil
+    end
+
+    def empty_connection_settings?
+      remote_broker_uri.blank? && @cluster.blank?
+    end
+
+    def cluster
+      @cluster ||= Acapi::ClusterSettings.new
+      yield @cluster if block_given?
+      @cluster
+    end
+
+    def to_connection_settings
+      raise ::Acapi::Errors::RemoteConnectionUnspecifiedError.new("No remote broker connection specified") if empty_connection_settings?
+      @connection_settings_hash ||= encode_connection_settings
+    end
+
+    def encode_connection_settings
+      if @cluster.blank?
+        uri = URI.parse(remote_broker_uri)
+        port_value = uri.port.blank? ? 5672 : uri.port
+        user_value = uri.user.blank? ? "guest" : uri.user
+        password_value = uri.password.blank? ? "guest" : uri.password
+        {
+          :host => uri.host,
+          :port => port_value,
+          :username => user_value,
+          :password => password_value,
+          :heartbeat => 10
+        }
+      else
+        cluster.to_connection_settings
+      end
+    end
   end
 end
 
@@ -28,12 +89,12 @@ module Railties
     class AmqpConfigurationSettings < Rails::Railtie
       config.after_initialize do |app|
         app_id = Rails.application.config.acapi.app_id
-        setting = Rails.application.config.acapi.remote_broker_uri
+        setting = Rails.application.config.acapi
         r_exchange = Rails.application.config.acapi.remote_request_exchange
-        if !setting
+        if Rails.application.config.acapi.empty_connection_settings?
           disable_requestor
         else
-          boot_requestor(app_id, setting, r_exchange)
+          boot_requestor(app_id, setting.to_connection_settings, r_exchange)
         end
       end
 

lib/acapi/requestor.rb

@@ -32,15 +32,14 @@ def request(req_name, payload,timeout=1)
       end
 
       def open_connection_for_request
-        if !@connection
-          @connection = Bunny.new(@uri, :heartbeat => 15)
-          @connection.start
-        end
+        return if @connection.present? && @connection.connected?
+        @connection = Bunny.new(@uri)
+        @connection.start
       end
 
       def reconnect!
         disconnect!
-        @connection = Bunny.new(@uri, :heartbeat => 15)
+        @connection = Bunny.new(@uri)
         @connection.start
       end
 

lib/acapi/sneakers_extensions.rb

@@ -17,7 +17,7 @@ def with_confirmed_channel
         begin
           chan.confirm_select
           yield chan
-          chan.wait_for_confirms
+          chan.wait_for_confirms || raise(Acapi::Errors::PublishConfirmationFailedError, "message publication could not be confirmed")
         ensure
           chan.close
         end