diff --git a/.circleci/config.yml b/.circleci/config.yml index 24ef51e..50ba9dd 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -6,6 +6,9 @@ jobs: docker_image: type: string description: "The Ruby or JRuby Docker image to test against" + gemfile: + type: string + description: "The Appraisal gemfile (Rails version) to test against" docker: # 1. The Primary Container (where your code actually runs) @@ -13,15 +16,19 @@ jobs: environment: JRUBY_OPTS: "-J-Xmx1024m" RAILS_ENV: test - # Tell your app where to find RabbitMQ (if your app uses this ENV var) + # Select the Rails version under test via the Appraisal gemfile. + BUNDLE_GEMFILE: << parameters.gemfile >> + # Tell the suite where to find RabbitMQ. RABBITMQ_URL: "amqp://guest:guest@localhost:5672" + RABBITMQ_HOST: "localhost" + RABBITMQ_PORT: "5672" - # 2. The Service Container (runs in the background) + # 2. The Service Container (runs in the background). + # NOTE: action_subscriber declares transient (non-durable) queues by default, + # which RabbitMQ 4.x denies out of the box. rabbitmq:3.12 still permits them. + # If/when moving to a 4.x image, permit the deprecated feature via config: + # deprecated_features.permit.transient_nonexcl_queues = true - image: rabbitmq:3.12-management - # If you need the management UI for debugging, use `rabbitmq:3-management` instead - # environment: - # RABBITMQ_DEFAULT_USER: guest - # RABBITMQ_DEFAULT_PASS: guest working_directory: ~/project @@ -35,24 +42,49 @@ jobs: sudo apt-get update && sudo apt-get install -y build-essential git fi - checkout - # Note: We added the docker_image parameter to the cache key - # so MRI and JRuby gems don't conflict. + # Cache key includes the Ruby image + the specific appraisal gemfile + the gemspec + # + the Appraisals file, so MRI/JRuby and each Rails version get independent caches + # and a dependency change in either file busts them. (All lockfiles are gitignored, + # so we key on committed sources instead.) - restore_cache: keys: - - v1-gems-<< parameters.docker_image >>-{{ checksum "Gemfile.lock" }} - - v1-gems-<< parameters.docker_image >>- + - v4-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} + - v4-gems-<< parameters.docker_image >>-<< parameters.gemfile >>- + + # gemfiles/ is gitignored and generated here, so it does not exist at checkout. + # + # Two things this step has to get right: + # 1. BUNDLE_GEMFILE is set job-wide to the target appraisal gemfile, which does + # not exist yet. Override it to the root Gemfile or appraisal tries to read + # the very file it is about to write. + # 2. appraisal runs under bundler, so the root Gemfile must be *installed* + # first -- otherwise it aborts with "Could not find gem ... in locally + # installed gems". The two bundles differ only in their Rails pins and share + # vendor/bundle, so the second install below is mostly a no-op. + - run: + name: Generate Appraisal Gemfiles + command: | + gem install bundler appraisal + bundle config set --local path 'vendor/bundle' + BUNDLE_GEMFILE=Gemfile bundle install --jobs=4 --retry=3 + BUNDLE_GEMFILE=Gemfile bundle exec appraisal generate + ls -1 gemfiles/ - run: name: Install Ruby Dependencies command: | - gem install bundler bundle config set --local path 'vendor/bundle' bundle install --jobs=4 --retry=3 + # Two paths: `bundle config --local` is relative to the directory holding + # BUNDLE_GEMFILE, so the root bundle lands in ./vendor/bundle while the + # appraisal bundle lands in ./gemfiles/vendor/bundle. Caching only the first + # would silently reinstall the gems the tests actually run against. - save_cache: paths: - ./vendor/bundle - key: v1-gems-<< parameters.docker_image >>-{{ checksum "Gemfile.lock" }} + - ./gemfiles/vendor/bundle + key: v4-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} # Wait for RabbitMQ to be ready before running tests. # Service containers can sometimes take a few seconds to boot up. @@ -77,14 +109,35 @@ jobs: workflows: version: 2 - ruby_compatibility_matrix: + ruby_rails_compatibility_matrix: jobs: + # Rails 6.1 - 7.2 run on every supported Ruby. Rails 6.1/7.0/7.1 need the + # default-gem shims on Ruby >= 3.4 (see Appraisals); 7.2 needs Ruby >= 3.1. - build_and_test: - name: test-<< matrix.docker_image >> + name: test-<< matrix.docker_image >>-<< matrix.gemfile >> matrix: parameters: docker_image: - "cimg/ruby:3.1" - "cimg/ruby:3.4" - "jruby:9.4" - - "jruby:10.0" \ No newline at end of file + - "jruby:10.0" + gemfile: + - "gemfiles/rails_6.1.gemfile" + - "gemfiles/rails_7.0.gemfile" + - "gemfiles/rails_7.1.gemfile" + - "gemfiles/rails_7.2.gemfile" + + # Rails 8.0/8.1 require Ruby >= 3.2, so they get their own matrix rather + # than excludes against the Ruby 3.1-class images. jruby:9.4 targets Ruby + # 3.1 compatibility and is therefore not eligible here either. + - build_and_test: + name: test-<< matrix.docker_image >>-<< matrix.gemfile >> + matrix: + parameters: + docker_image: + - "cimg/ruby:3.4" + - "jruby:10.0" + gemfile: + - "gemfiles/rails_8.0.gemfile" + - "gemfiles/rails_8.1.gemfile" diff --git a/.gitignore b/.gitignore index 88db647..f0f8b8f 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,11 @@ tmp Gemfile.lock +# Appraisal output. Both the .gemfile stubs and their locks are generated from +# the Appraisals file -- CI regenerates them, and locally you run: +# bundle exec appraisal generate +gemfiles/ + # YARD artifacts .yardoc _yardoc diff --git a/Appraisals b/Appraisals new file mode 100644 index 0000000..a9f4123 --- /dev/null +++ b/Appraisals @@ -0,0 +1,51 @@ +# Appraisal matrix for the Rails/ActiveSupport versions action_subscriber supports. +# +# Notes on Ruby compatibility when running this matrix in CI: +# * rails 6.1 / 7.0 / 7.1 run on Ruby >= 2.7 (and on Ruby 3.4 with the +# default-gem shims added below, since logger/mutex_m/bigdecimal/drb/base64 +# were removed from the default gem set). +# * rails 7.2 requires Ruby >= 3.1. +# * rails 8.0 / 8.1 require Ruby >= 3.2. +# Pair each gemfile with a compatible Ruby in the CI matrix. + +# Shims required by ActiveSupport < 7.1 on Ruby >= 3.4 (default gems removed). +older_rails_shims = proc do + gem "logger" + gem "mutex_m" + gem "bigdecimal" + gem "drb" + gem "base64" + gem "benchmark" +end + +appraise "rails-6.1" do + instance_exec(&older_rails_shims) + gem "activesupport", "~> 6.1.0" + gem "activerecord", "~> 6.1.0" +end + +appraise "rails-7.0" do + instance_exec(&older_rails_shims) + gem "activesupport", "~> 7.0.0" + gem "activerecord", "~> 7.0.0" +end + +appraise "rails-7.1" do + gem "activesupport", "~> 7.1.0" + gem "activerecord", "~> 7.1.0" +end + +appraise "rails-7.2" do + gem "activesupport", "~> 7.2.0" + gem "activerecord", "~> 7.2.0" +end + +appraise "rails-8.0" do + gem "activesupport", "~> 8.0.0" + gem "activerecord", "~> 8.0.0" +end + +appraise "rails-8.1" do + gem "activesupport", "~> 8.1.0" + gem "activerecord", "~> 8.1.0" +end diff --git a/README.md b/README.md index c9b35d5..09ea620 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,86 @@ end That will give you a similar behavior to the old `--mode=pop` where messages polled from the server, but with reduced latency. +Queue Types +----------- + +Set the queue type globally, or per route: + +```ruby +::ActionSubscriber.configure do |config| + config.queue_type = :quorum +end + +::ActionSubscriber.draw_routes do + route UserSubscriber, :created, :queue_type => :quorum + route AuditSubscriber, :created, :queue_type => :broker_default +end +``` + +| Value | `x-queue-type` sent | +| --- | --- | +| `nil` (default), or `:broker_default` | *not sent* — the broker applies its own `default_queue_type` | +| `:classic` | `classic` | +| `:quorum` | `quorum` | +| `:stream` | `stream` | + +The default declares a queue without expressing an opinion, which lets an +operator move a vhost onto quorum queues with a broker policy instead of a code +change. `:broker_default` is accepted as a more readable spelling of `nil`; both +normalize to `nil`, and `config.queue_type` always reads back as `nil` or one of +the three type symbols. + +`:quorum` and `:stream` queues only exist as durable queues, so those two values +force `:durable => true` on the route regardless of what you pass. + +Invalid values raise an `ArgumentError` at the point they are assigned, rather +than later when routes are drawn or a queue is declared. + +> Note: a queue's type is fixed at declaration. Changing this setting will not +> convert an existing queue — the queue has to be deleted and redeclared, and +> redeclaring an existing queue with a conflicting type fails with +> `PRECONDITION_FAILED`. + +### Breaking change on JRuby + +Prior to this setting the two drivers disagreed. `march_hare` defaults its +`:type` option to `classic` and so injected `x-queue-type: classic` on every +queue it declared, while `bunny` sent no argument at all. ActionSubscriber now +passes `:type` explicitly on both drivers and defaults to `nil`, so neither +platform sends `x-queue-type`. + +**MRI behavior is unchanged. On JRuby, newly declared queues change from +`classic` to whatever the broker defaults to.** Set `config.queue_type = :classic` +to keep the previous JRuby behavior. + +The reason this matters beyond new queues: because queue type is fixed at +declaration, an existing `classic` queue is now *redeclared* without +`x-queue-type`. That is harmless on a vhost whose `default_queue_type` is +classic, since the broker resolves to the same type. It fails with +`PRECONDITION_FAILED` on a vhost whose default is `quorum` or `stream`. Before +upgrading a JRuby deployment, audit the `default_queue_type` of every vhost it +connects to: + +``` +rabbitmqctl list_vhosts name default_queue_type +``` + +If any are non-classic, set `config.queue_type = :classic` before rolling out. + +### Known limitation: retry queues + +`ActionSubscriber::MessageRetry` declares its `*.retry_*` queues using the +**global** `config.queue_type`, not the type of the route that produced the +message. A route that opts into `:quorum` while the global setting is left at +the default will dead-letter into a retry queue of a different type. + +If you rely on per-route queue types and on retries, set `config.queue_type` to +match rather than setting it per route. + +Note also that retry queues carry `x-message-ttl` and `x-dead-letter-exchange`, +which streams do not support — so a global `config.queue_type = :stream` will +make every retry declaration fail. + Supported Message Types ----------------- ActionSubscriber support JSON and plain text out of the box, but you can easily @@ -135,6 +215,7 @@ Other configuration options include : * config.network_recovery_interval - reconnection interval for TCP connection failures (default 1) * config.password - RabbitMQ password (default "guest") * config.prefetch - number of messages to hold in the local queue in subscriber mode +* config.queue_type - default queue type for all routes: `nil` (default, defers to the broker), `:classic`, `:quorum` or `:stream` * config.resubscribe_on_consumer_cancellation - resubscribe when the consumer is cancelled (queue deleted or cluster fails, default true) * config.seconds_to_wait_for_graceful_shutdown - time to wait before force stopping server after shutdown signal * config.threadpool_size - set the number of threads available to action_subscriber @@ -241,3 +322,25 @@ $ cd action_subscriber $ bundle install $ bundle exec rspec ``` + +### Testing against multiple Rails versions + +The supported Rails versions are declared in `Appraisals`. The `gemfiles/` +directory is **generated, not committed** — it is gitignored, and CI regenerates +it on every run. To create it locally: + +``` +$ bundle exec appraisal generate # writes gemfiles/*.gemfile +$ bundle exec appraisal install # resolves a lockfile for each +``` + +Then run the suite against one version, or all of them: + +``` +$ BUNDLE_GEMFILE=gemfiles/rails_8.0.gemfile bundle exec rspec +$ bundle exec appraisal rspec +``` + +Re-run `appraisal generate` after editing `Appraisals`. Note that Rails 7.2 +requires Ruby >= 3.1 and Rails 8.0/8.1 require Ruby >= 3.2, so those gemfiles +will not resolve on older interpreters. diff --git a/Rakefile b/Rakefile index 950a17e..9c3a189 100644 --- a/Rakefile +++ b/Rakefile @@ -4,5 +4,14 @@ require "rspec/core/rake_task" desc "Run specs" RSpec::Core::RakeTask.new(:spec) +# Appraisal wires up per-Rails-version tasks (rake appraisal:rails-8.1 spec, etc.) +# when the appraisal gem is available. It is only a development dependency, so we +# guard the require to keep the Rakefile usable without it (e.g. from an installed gem). +begin + require "appraisal" +rescue LoadError + # appraisal not installed; per-version tasks are unavailable +end + desc "Run specs (default)" task :default => :spec diff --git a/action_subscriber.gemspec b/action_subscriber.gemspec index b2cf28c..fcdf0b5 100644 --- a/action_subscriber.gemspec +++ b/action_subscriber.gemspec @@ -32,8 +32,9 @@ Gem::Specification.new do |spec| spec.add_dependency "middleware" spec.add_dependency "thor" - spec.add_development_dependency "active_publisher", "1.6.0.pre1" + spec.add_development_dependency "active_publisher", "1.6.0" spec.add_development_dependency "activerecord", ">= 6.0" + spec.add_development_dependency "appraisal", "~> 2.5" spec.add_development_dependency "bundler" spec.add_development_dependency "pry-nav" spec.add_development_dependency "rabbitmq_http_api_client", "~> 1.15.0" diff --git a/changelog.md b/changelog.md index 108c0b3..8612ad5 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,25 @@ ### Changelog +### 6.0.0 - August 6, 2026 + +**Breaking change on JRuby.** `march_hare` defaults its `:type` option to +`classic` and so was injecting `x-queue-type: classic` on every queue it +declared, while `bunny` sent no argument at all. Both drivers are now passed +`:type` explicitly, defaulting to `nil`, so neither sends `x-queue-type` and the +broker's own `default_queue_type` applies. + +MRI behavior is unchanged. On JRuby, newly declared queues change from `classic` +to whatever the broker defaults to. Set `config.queue_type = :classic` to retain +the previous JRuby behavior. Note that queue type is fixed at declaration: +redeclaring an existing queue with a conflicting type fails with +`PRECONDITION_FAILED`, so audit any vhost whose `default_queue_type` is not +classic before upgrading. + +Added a first party `queue_type` setting behind that change, configurable +globally (`config.queue_type`) or per route (`:queue_type => ...`). Values are +`nil` (default), `:classic`, `:quorum` and `:stream`; `:broker_default` is +accepted as a readable alias for `nil`. `:quorum` and `:stream` force the route +to be durable. Invalid values raise where they are assigned. ### 5.4.0 - April 10, 2026 Added Ruby 3.4 / JRuby 10 support. \ No newline at end of file diff --git a/lib/action_subscriber/bunny/subscriber.rb b/lib/action_subscriber/bunny/subscriber.rb index 1957a1a..4ff037f 100644 --- a/lib/action_subscriber/bunny/subscriber.rb +++ b/lib/action_subscriber/bunny/subscriber.rb @@ -76,7 +76,7 @@ def start_subscriber_for_subscription(subscription) def setup_queue(route) channel = ::ActionSubscriber::RabbitConnection.with_connection{|connection| connection.create_channel(nil, 1) } exchange = channel.topic(route.exchange) - queue = channel.queue(route.queue, :durable => route.durable) + queue = channel.queue(route.queue, :durable => route.durable, :type => route.driver_queue_type) queue.bind(exchange, :routing_key => route.routing_key) queue end diff --git a/lib/action_subscriber/configuration.rb b/lib/action_subscriber/configuration.rb index 9f9a8d2..0ab7ea0 100644 --- a/lib/action_subscriber/configuration.rb +++ b/lib/action_subscriber/configuration.rb @@ -1,4 +1,5 @@ require "yaml" +require "action_subscriber/queue_type" require "action_subscriber/uri" module ActionSubscriber @@ -28,6 +29,10 @@ class Configuration :verify_peer, :virtual_host + # Written through QueueType.normalize so a typo raises where it was set + # rather than at route-draw time. nil means "defer to the broker". + attr_reader :queue_type + CONFIGURATION_MUTEX = ::Mutex.new NETWORK_RECOVERY_INTERVAL = 1.freeze @@ -43,6 +48,7 @@ class Configuration :password => "guest", :port => 5672, :prefetch => 2, + :queue_type => nil, :resubscribe_on_consumer_cancellation => true, :seconds_to_wait_for_graceful_shutdown => 30, :threadpool_size => 8, @@ -144,6 +150,10 @@ def middleware @middleware ||= Middleware.initialize_stack end + def queue_type=(value) + @queue_type = ::ActionSubscriber::QueueType.normalize(value) + end + def inspect inspection_string = <<-INSPECT.strip_heredoc Rabbit Hosts: #{hosts} diff --git a/lib/action_subscriber/march_hare/subscriber.rb b/lib/action_subscriber/march_hare/subscriber.rb index 06a89f4..45038d4 100644 --- a/lib/action_subscriber/march_hare/subscriber.rb +++ b/lib/action_subscriber/march_hare/subscriber.rb @@ -75,7 +75,9 @@ def start_subscriber_for_subscription(subscription) def setup_queue(route) channel = ::ActionSubscriber::RabbitConnection.with_connection{|connection| connection.create_channel } exchange = channel.topic(route.exchange) - queue = channel.queue(route.queue, :durable => route.durable) + # :type must be passed even when nil -- omitting it declares a classic + # queue here, unlike bunny. See ActionSubscriber::QueueType. + queue = channel.queue(route.queue, :durable => route.durable, :type => route.driver_queue_type) queue.bind(exchange, :routing_key => route.routing_key) queue end diff --git a/lib/action_subscriber/message_retry.rb b/lib/action_subscriber/message_retry.rb index ab95914..6a81fed 100644 --- a/lib/action_subscriber/message_retry.rb +++ b/lib/action_subscriber/message_retry.rb @@ -51,7 +51,8 @@ def self.with_exchange(env, ttl, retry_queue_name) channel.confirm_select # an empty string is the default exchange [see bunny docs](http://rubybunny.info/articles/exchanges.html#default_exchange) exchange = channel.topic("") - queue = channel.queue(retry_queue_name, :arguments => {"x-dead-letter-exchange" => "", "x-message-ttl" => ttl, "x-dead-letter-routing-key" => env.queue}) + queue_type = ::ActionSubscriber::QueueType.driver_option(::ActionSubscriber.config.queue_type) + queue = channel.queue(retry_queue_name, :type => queue_type, :arguments => {"x-dead-letter-exchange" => "", "x-message-ttl" => ttl, "x-dead-letter-routing-key" => env.queue}) yield(exchange) channel.wait_for_confirms end diff --git a/lib/action_subscriber/queue_type.rb b/lib/action_subscriber/queue_type.rb new file mode 100644 index 0000000..e9c3872 --- /dev/null +++ b/lib/action_subscriber/queue_type.rb @@ -0,0 +1,46 @@ +module ActionSubscriber + # Normalizes the `queue_type` setting into the value the underlying driver + # expects for its `:type` option. + # + # nil is the canonical "let the broker decide" value: it leaves `x-queue-type` + # off the wire so RabbitMQ applies its own `default_queue_type`. + # + # march_hare reads the driver option with + # `@options.fetch(:type, ... Types::CLASSIC)`, and `fetch` only falls back when + # the key is *absent*. So omitting `:type` silently declares a classic queue, + # while passing an explicit nil is what actually suppresses the argument. + module QueueType + # An explicit, readable alias for nil on input. Normalizes away to nil. + BROKER_DEFAULT = :broker_default + + SUPPORTED = [:classic, :quorum, :stream].freeze + + # Queue types that RabbitMQ only supports as durable queues. + ALWAYS_DURABLE = [:quorum, :stream].freeze + + def self.normalize(value) + queue_type = value.to_s.strip.downcase + return nil if queue_type.empty? || queue_type == BROKER_DEFAULT.to_s + + queue_type = queue_type.to_sym + unless SUPPORTED.include?(queue_type) + raise ::ArgumentError, + "unsupported queue_type #{value.inspect}, supported types are: #{SUPPORTED.join(', ')} " \ + "(or nil / :#{BROKER_DEFAULT} to defer to the broker)" + end + + queue_type + end + + # The value to hand the driver's `:type` option. Expects a normalized type. + def self.driver_option(queue_type) + return nil if queue_type.nil? + queue_type.to_s + end + + # Expects a normalized type. + def self.always_durable?(queue_type) + ALWAYS_DURABLE.include?(queue_type) + end + end +end diff --git a/lib/action_subscriber/route.rb b/lib/action_subscriber/route.rb index 9d25078..0a92598 100644 --- a/lib/action_subscriber/route.rb +++ b/lib/action_subscriber/route.rb @@ -2,10 +2,12 @@ module ActionSubscriber class Route attr_reader :acknowledgements, :action, + :driver_queue_type, :durable, :exchange, :prefetch, :queue, + :queue_type, :routing_key, :subscriber, :threadpool_name @@ -13,7 +15,16 @@ class Route def initialize(attributes) @acknowledgements = attributes.fetch(:acknowledgements) @action = attributes.fetch(:action) - @durable = attributes.fetch(:durable) + durable = attributes.fetch(:durable) + # Falls back to the global setting when a route does not name a type, the + # same way :prefetch does. nil means "defer to the broker". + @queue_type = ::ActionSubscriber::QueueType.normalize( + attributes.fetch(:queue_type) { ::ActionSubscriber.config.queue_type } + ) + @driver_queue_type = ::ActionSubscriber::QueueType.driver_option(@queue_type) + # Quorum and stream queues only exist as durable queues, so the broker + # rejects them otherwise. march_hare already forces this internally. + @durable = ::ActionSubscriber::QueueType.always_durable?(@queue_type) || durable @exchange = attributes.fetch(:exchange).to_s @prefetch = attributes.fetch(:prefetch) { ::ActionSubscriber.config.prefetch } @queue = attributes.fetch(:queue) diff --git a/lib/action_subscriber/version.rb b/lib/action_subscriber/version.rb index cdfa208..4b0a00c 100644 --- a/lib/action_subscriber/version.rb +++ b/lib/action_subscriber/version.rb @@ -1,3 +1,3 @@ module ActionSubscriber - VERSION = "5.4.0" + VERSION = "6.0.0" end diff --git a/spec/lib/action_subscriber/queue_type_spec.rb b/spec/lib/action_subscriber/queue_type_spec.rb new file mode 100644 index 0000000..ab43500 --- /dev/null +++ b/spec/lib/action_subscriber/queue_type_spec.rb @@ -0,0 +1,96 @@ +describe ActionSubscriber::QueueType do + describe ".normalize" do + it "treats nil and blank strings as the broker default" do + expect(described_class.normalize(nil)).to be_nil + expect(described_class.normalize("")).to be_nil + expect(described_class.normalize(" ")).to be_nil + end + + it "treats :broker_default as an alias for nil" do + expect(described_class.normalize(:broker_default)).to be_nil + expect(described_class.normalize("broker_default")).to be_nil + end + + it "accepts strings and symbols for the supported types" do + expect(described_class.normalize("classic")).to eq(:classic) + expect(described_class.normalize(:quorum)).to eq(:quorum) + expect(described_class.normalize("STREAM")).to eq(:stream) + end + + it "raises on an unsupported type" do + expect { described_class.normalize(:mirrored) }.to raise_error(ArgumentError, /unsupported queue_type/) + end + end + + describe ".driver_option" do + # nil is what keeps x-queue-type off the wire in both bunny and march_hare. + it "is nil for the broker default" do + expect(described_class.driver_option(nil)).to be_nil + end + + it "is the type name for an explicit type" do + expect(described_class.driver_option(:quorum)).to eq("quorum") + expect(described_class.driver_option(:classic)).to eq("classic") + end + end + + describe ".always_durable?" do + it "is true for quorum and stream queues" do + expect(described_class.always_durable?(:quorum)).to eq(true) + expect(described_class.always_durable?(:stream)).to eq(true) + end + + it "is false for classic and broker default queues" do + expect(described_class.always_durable?(:classic)).to eq(false) + expect(described_class.always_durable?(nil)).to eq(false) + end + end + + describe "configuration" do + around do |example| + original = ActionSubscriber.config.queue_type + example.run + ActionSubscriber.config.queue_type = original + end + + it "defaults to nil" do + expect(ActionSubscriber.config.queue_type).to be_nil + end + + it "normalizes on assignment so readers always see a symbol or nil" do + ActionSubscriber.config.queue_type = "quorum" + expect(ActionSubscriber.config.queue_type).to eq(:quorum) + + ActionSubscriber.config.queue_type = :broker_default + expect(ActionSubscriber.config.queue_type).to be_nil + end + + it "raises at the point the bad value is set" do + expect { ActionSubscriber.config.queue_type = "qourum" }.to raise_error(ArgumentError, /unsupported queue_type/) + end + end + + # The behavior this whole module exists for: march_hare reads its :type option + # with fetch(:type, ... CLASSIC), so omitting the key declares a classic queue + # while passing an explicit nil leaves x-queue-type off the wire. + if ::RUBY_PLATFORM == "java" + describe "march_hare integration" do + def arguments_for(type) + ::MarchHare::Queue.new(nil, "test.queue", :durable => false, :type => type).arguments + end + + it "sends no x-queue-type when the driver option is nil" do + expect(arguments_for(described_class.driver_option(nil))).to eq({}) + end + + it "sends x-queue-type when a type is named" do + expect(arguments_for(described_class.driver_option(:quorum))).to eq("x-queue-type" => "quorum") + end + + it "would send classic if the :type key were omitted entirely" do + omitted = ::MarchHare::Queue.new(nil, "test.queue", :durable => false).arguments + expect(omitted).to eq("x-queue-type" => "classic") + end + end + end +end diff --git a/spec/lib/action_subscriber/router_spec.rb b/spec/lib/action_subscriber/router_spec.rb index 24eeee8..ac35676 100644 --- a/spec/lib/action_subscriber/router_spec.rb +++ b/spec/lib/action_subscriber/router_spec.rb @@ -57,6 +57,53 @@ class FakeSubscriber; end expect(routes.first.queue).to eq("alice.fake.foo") end + it "defers to the broker by default" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo + end + + expect(routes.first.queue_type).to be_nil + expect(routes.first.driver_queue_type).to be_nil + end + + it "accepts :broker_default as an explicit alias for nil" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo, :queue_type => :broker_default + end + + expect(routes.first.queue_type).to be_nil + expect(routes.first.driver_queue_type).to be_nil + end + + it "can specify a queue type" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo, :queue_type => :classic + end + + expect(routes.first.queue_type).to eq(:classic) + expect(routes.first.driver_queue_type).to eq("classic") + expect(routes.first.durable).to eq(false) + end + + it "forces quorum queues to be durable" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo, :queue_type => :quorum + end + + expect(routes.first.queue_type).to eq(:quorum) + expect(routes.first.durable).to eq(true) + end + + it "inherits the queue type from the global configuration" do + allow(ActionSubscriber.config).to receive(:queue_type).and_return(:quorum) + + routes = described_class.draw_routes do + route FakeSubscriber, :foo + end + + expect(routes.first.queue_type).to eq(:quorum) + end + it "can specify a queue is durable" do routes = described_class.draw_routes do route FakeSubscriber, :foo, :durable => true diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index cd6b9a1..5e66089 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -18,6 +18,7 @@ # Require spec support files require 'support/user_subscriber' +require 'support/rabbitmq' require 'action_subscriber/rspec' # Silence the Logger @@ -30,6 +31,12 @@ mocks.verify_partial_doubles = true end + # Fail fast with a clear message (rather than a flurry of Bunny reconnect warnings) + # if the broker isn't up yet when the integration suite starts. + config.before(:suite) do + RabbitMQTestHelper.wait_for_rabbitmq! if RSpec.world.filtered_examples.values.flatten.any? { |ex| ex.metadata[:integration] } + end + config.before(:each, :integration => true) do $messages = Set.new draw_routes diff --git a/spec/support/rabbitmq.rb b/spec/support/rabbitmq.rb new file mode 100644 index 0000000..98353a0 --- /dev/null +++ b/spec/support/rabbitmq.rb @@ -0,0 +1,52 @@ +require "socket" + +# Helpers for running the suite against a real RabbitMQ broker. +# +# The integration specs talk to a live broker (the same approach used in CI, where a +# `rabbitmq` service container is started alongside the test job). Locally you can point +# at any running broker via RABBITMQ_URL / the standard host+port; by default we assume +# localhost:5672. +# +# NOTE: action_subscriber defaults to non-durable ("transient") queues. RabbitMQ 4.x +# denies transient non-exclusive queues by default, so a 4.x broker used for the suite +# must permit the deprecated feature: +# +# # rabbitmq.conf +# deprecated_features.permit.transient_nonexcl_queues = true +# +# The rabbitmq:3.12 image used in CI still allows them out of the box. See the phantom +# queue triage doc for why the production recommendation is to move to durable topology. +module RabbitMQTestHelper + module_function + + def host + ENV.fetch("RABBITMQ_HOST", "localhost") + end + + def port + Integer(ENV.fetch("RABBITMQ_PORT", "5672")) + end + + # Block until the broker's AMQP port accepts a TCP connection, or raise after `timeout` + # seconds. Keeps the suite from failing with confusing connection errors when the broker + # is still booting (common in CI service containers). + def wait_for_rabbitmq!(timeout: Integer(ENV.fetch("RABBITMQ_WAIT_TIMEOUT", "30"))) + deadline = ::Time.now + timeout + last_error = nil + loop do + begin + ::Socket.tcp(host, port, connect_timeout: 1) { |sock| sock.close } + return true + rescue ::StandardError => e + last_error = e + end + + if ::Time.now >= deadline + raise "RabbitMQ was not reachable at #{host}:#{port} within #{timeout}s " \ + "(last error: #{last_error.class}: #{last_error.message}). " \ + "Start a broker (see spec/support/rabbitmq.rb) before running the integration suite." + end + sleep 0.5 + end + end +end