From 11d22f8e22e0c2252887681d85db2ef975139829 Mon Sep 17 00:00:00 2001 From: fdotoliveira Date: Mon, 27 Apr 2026 13:52:10 -0300 Subject: [PATCH] feat: support shoryuken processing groups --- README.md | 31 +++++++ lib/aws/sqs/configurator/queue.rb | 2 +- lib/turtle.rb | 8 +- lib/turtle/group.rb | 45 ++++++++++ .../configs/with_all_queues_grouped.yml | 28 +++++++ .../configs/with_shoryuken_groups.yml | 36 ++++++++ spec/turtle/group_spec.rb | 82 +++++++++++++++++++ spec/turtle_spec.rb | 45 +++++++++- 8 files changed, 271 insertions(+), 6 deletions(-) create mode 100644 lib/turtle/group.rb create mode 100644 spec/fixtures/configs/with_all_queues_grouped.yml create mode 100644 spec/fixtures/configs/with_shoryuken_groups.yml create mode 100644 spec/turtle/group_spec.rb diff --git a/README.md b/README.md index fa5bc68..e23afc3 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,37 @@ Turtle.shoryuken_queues_priorities(priority: 3) # ["macaw_linquetab_perform_subscription_events", 3]] ``` +### Processing groups for shoryuken +To use [Processing Groups](https://github.com/ruby-shoryuken/shoryuken/wiki/Processing-Groups), add `groups:` to your `aws-sqs-configurator` config: + +```yml +queues: + - name: 'orders_queue' + metadata: + priority: 2 + - name: 'payments_queue' + metadata: + priority: 1 + +groups: + critical: + concurrency: 1 + queues: + - orders_queue + - payments_queue +``` + +```ruby +Turtle.shoryuken_groups +# => "{\"critical\":{\"concurrency\":1,\"queues\":[[\"app_production_orders_queue\",2],[\"app_production_payments_queue\",1]]}}" +``` + +In `config/shoryuken.yml`: +```yml +queues: <%= Turtle.shoryuken_queues_priorities %> +groups: <%= Turtle.shoryuken_groups %> +``` + ### Name for ```ruby # Turtle.name_for(type, name, options) diff --git a/lib/aws/sqs/configurator/queue.rb b/lib/aws/sqs/configurator/queue.rb index fd6b685..0da6554 100644 --- a/lib/aws/sqs/configurator/queue.rb +++ b/lib/aws/sqs/configurator/queue.rb @@ -126,7 +126,7 @@ def build_dead_letter!(options) def dead_letter_options(options) options.merge(dead_letter_queue: false, topics: [], suffix: [@suffix, @dead_letter_queue_suffix] - .compact.join('_')) + .compact.join('_')) end def build_attributes! diff --git a/lib/turtle.rb b/lib/turtle.rb index f06f66d..b7fb538 100644 --- a/lib/turtle.rb +++ b/lib/turtle.rb @@ -1,6 +1,7 @@ # frozen_string_literal: true require 'turtle/version' +require 'turtle/group' require 'turtle/queue' require 'turtle/topic' require 'turtle/logger' @@ -14,7 +15,12 @@ module Turtle class << self def shoryuken_queues_priorities(options = nil) - Queue.shoryuken_priorities(options) + queues_in_groups = Group.to_h.values.flat_map { |attrs| attrs[:queues].map { |name, _| name } } + Queue.shoryuken_priorities(options).reject { |name, _| queues_in_groups.include?(name) } + end + + def shoryuken_groups + Group.to_json end def delayed_job_queue_attributes diff --git a/lib/turtle/group.rb b/lib/turtle/group.rb new file mode 100644 index 0000000..8660ce1 --- /dev/null +++ b/lib/turtle/group.rb @@ -0,0 +1,45 @@ +# frozen_string_literal: true + +require 'aws/sqs/configurator' +require 'yaml' +require 'erb' +require 'turtle/logger' + +module Turtle + class Group + class << self + def to_h + @to_h ||= begin + queues = AWS::SQS::Configurator.queues! + config_groups.each_with_object({}) do |(name, attrs), groups| + groups[name] = attrs.except(:queues).merge(queues: queue_list(Array(attrs[:queues]), queues)) + queue_names = groups[name][:queues].map(&:first).join(', ') + Logger.info("Group added: #{name} concurrency: #{groups[name][:concurrency]} queues: #{queue_names}") + end + end + end + + def to_json(*) + to_h.to_json + end + + private + + def config_groups + merged = config_files.select { |f| File.exist?(f) }.each_with_object({}) do |file, groups| + groups.merge!(YAML.safe_load(ERB.new(File.read(file)).result).to_h.fetch('groups', {})) + end + merged.transform_values { |attrs| attrs.transform_keys(&:to_sym) } + end + + def config_files + Dir[AWS::SQS::Configurator::Reader::DIR_FILES] << AWS::SQS::Configurator::Reader::MAIN_FILE + end + + def queue_list(names, queues) + queues.select { |q| names.include?(q.name) } + .map { |q| [q.name_formatted, q.metadata[:priority] || 1] } + end + end + end +end diff --git a/spec/fixtures/configs/with_all_queues_grouped.yml b/spec/fixtures/configs/with_all_queues_grouped.yml new file mode 100644 index 0000000..6db3df3 --- /dev/null +++ b/spec/fixtures/configs/with_all_queues_grouped.yml @@ -0,0 +1,28 @@ +--- +default: + general: + region: 'us-east-1' + prefix: 'app_name' + environment: 'production' + queue: + dead_letter_queue: false +queues: + - name: 'grouped_solo' + metadata: + priority: 1 + - name: 'grouped_multi_a' + metadata: + priority: 2 + - name: 'grouped_multi_b' + metadata: + priority: 2 +groups: + solo_group: + concurrency: 1 + queues: + - grouped_solo + multi_group: + concurrency: 2 + queues: + - grouped_multi_a + - grouped_multi_b diff --git a/spec/fixtures/configs/with_shoryuken_groups.yml b/spec/fixtures/configs/with_shoryuken_groups.yml new file mode 100644 index 0000000..4743794 --- /dev/null +++ b/spec/fixtures/configs/with_shoryuken_groups.yml @@ -0,0 +1,36 @@ +--- +default: + general: + region: 'us-east-1' + prefix: 'app_name' + environment: 'production' + queue: + dead_letter_queue: false +queues: + - name: 'regular_a' + metadata: + priority: 5 + - name: 'regular_b' + metadata: + priority: 3 + - name: 'batch_solo' + metadata: + priority: 1 + - name: 'batch_multi_a' + metadata: + priority: 2 + - name: 'batch_multi_b' + metadata: + priority: 2 +groups: + batch_solo: + concurrency: 1 + queues: + - batch_solo + batch_multi: + concurrency: 3 + delay: 30 + polling_strategy: 'WeightedRoundRobin' # Default, declaration optional + queues: + - batch_multi_a + - batch_multi_b diff --git a/spec/turtle/group_spec.rb b/spec/turtle/group_spec.rb new file mode 100644 index 0000000..36abfbd --- /dev/null +++ b/spec/turtle/group_spec.rb @@ -0,0 +1,82 @@ +# frozen_string_literal: true + +RSpec.describe Turtle::Group, type: :model do + after { described_class.instance_variable_set(:@to_h, nil) } + + describe '.to_h' do + subject { described_class.to_h } + + context 'without groups section' do + before { stub_const('AWS::SQS::Configurator::Reader::MAIN_FILE', './spec/fixtures/configs/queues.yml') } + + it { is_expected.to eq({}) } + end + + context 'with groups section' do + before { stub_const('AWS::SQS::Configurator::Reader::MAIN_FILE', './spec/fixtures/configs/with_shoryuken_groups.yml') } + + it 'returns groups with resolved queues and attributes' do + is_expected.to eq( + 'batch_solo' => { concurrency: 1, queues: [['app_name_production_batch_solo', 1]] }, + 'batch_multi' => { + concurrency: 3, + delay: 30, + polling_strategy: 'WeightedRoundRobin', + queues: [['app_name_production_batch_multi_a', 2], ['app_name_production_batch_multi_b', 2]] + } + ) + end + end + + context 'with groups in DIR_FILES' do + before do + stub_const('AWS::SQS::Configurator::Reader::DIR_FILES', './spec/fixtures/configs/with_shoryuken_groups.yml') + stub_const('AWS::SQS::Configurator::Reader::MAIN_FILE', './spec/fixtures/configs/nonexistent.yml') + end + + it 'returns groups resolved from directory config files' do + is_expected.to eq( + 'batch_solo' => { concurrency: 1, queues: [['app_name_production_batch_solo', 1]] }, + 'batch_multi' => { + concurrency: 3, + delay: 30, + polling_strategy: 'WeightedRoundRobin', + queues: [['app_name_production_batch_multi_a', 2], ['app_name_production_batch_multi_b', 2]] + } + ) + end + end + + context 'with all queues grouped' do + before { stub_const('AWS::SQS::Configurator::Reader::MAIN_FILE', './spec/fixtures/configs/with_all_queues_grouped.yml') } + + it 'returns all groups with their attributes and queues' do + is_expected.to eq( + 'solo_group' => { concurrency: 1, queues: [['app_name_production_grouped_solo', 1]] }, + 'multi_group' => { concurrency: 2, queues: [['app_name_production_grouped_multi_a', 2], ['app_name_production_grouped_multi_b', 2]] } + ) + end + end + end + + describe '.to_h logging' do + before { stub_const('AWS::SQS::Configurator::Reader::MAIN_FILE', './spec/fixtures/configs/with_shoryuken_groups.yml') } + + it 'logs each group with name, concurrency and queues' do + expect(Turtle::Logger).to receive(:info).with('Group added: batch_solo concurrency: 1 queues: app_name_production_batch_solo') + expect(Turtle::Logger).to receive(:info).with('Group added: batch_multi concurrency: 3 queues: app_name_production_batch_multi_a, app_name_production_batch_multi_b') + described_class.to_h + end + end + + describe '.to_json' do + subject { described_class.to_json } + + before { stub_const('AWS::SQS::Configurator::Reader::MAIN_FILE', './spec/fixtures/configs/with_shoryuken_groups.yml') } + + it 'returns groups as a valid JSON string' do + parsed = JSON.parse(subject) + expect(parsed['batch_solo']).to eq('concurrency' => 1, 'queues' => [['app_name_production_batch_solo', 1]]) + end + end +end diff --git a/spec/turtle_spec.rb b/spec/turtle_spec.rb index 0a0dc3a..9b2e178 100644 --- a/spec/turtle_spec.rb +++ b/spec/turtle_spec.rb @@ -6,12 +6,49 @@ end describe '#shoryuken_queues_priorities' do - subject { described_class.shoryuken_queues_priorities({}) } + context 'call through' do + subject { described_class.shoryuken_queues_priorities({}) } - after { subject } + before { allow(described_class::Group).to receive(:to_h).and_return({}) } + after { subject } - it 'should call through Queue' do - expect(described_class::Queue).to receive(:shoryuken_priorities).once + it 'should call through Queue' do + expect(described_class::Queue).to receive(:shoryuken_priorities).once.and_return([]) + end + end + + context 'with grouped queues' do + subject { described_class.shoryuken_queues_priorities } + + before { stub_const('AWS::SQS::Configurator::Reader::MAIN_FILE', './spec/fixtures/configs/with_shoryuken_groups.yml') } + + it 'returns only queues not assigned to a group' do + is_expected.to eq([['app_name_production_regular_a', 5], ['app_name_production_regular_b', 3]]) + end + + it 'excludes grouped queues by name regardless of priority' do + allow(described_class::Queue).to receive(:shoryuken_priorities).and_return( + [['app_name_production_regular_a', 5], ['app_name_production_batch_solo', 99]] + ) + is_expected.to eq([['app_name_production_regular_a', 5]]) + end + end + end + + describe '#shoryuken_groups' do + subject { described_class.shoryuken_groups } + + before { stub_const('AWS::SQS::Configurator::Reader::MAIN_FILE', './spec/fixtures/configs/with_shoryuken_groups.yml') } + + it 'returns groups as JSON string ready for shoryuken.yml' do + parsed = JSON.parse(subject) + expect(parsed['batch_solo']).to eq('concurrency' => 1, 'queues' => [['app_name_production_batch_solo', 1]]) + end + + it 'produces no overlap with shoryuken_queues_priorities' do + queue_names = described_class.shoryuken_queues_priorities.map(&:first) + group_queue_names = JSON.parse(subject).values.flat_map { |g| g['queues'].map(&:first) } + expect(queue_names & group_queue_names).to be_empty end end