Add delay option into Shoryuken.add_group

.gitignore

@@ -25,3 +25,4 @@ shoryuken.yml
 *.log
 .env
 rubocop.html
+.byebug_history

lib/shoryuken/environment_loader.rb

@@ -130,7 +130,7 @@ def parse_queues
       end
 
       Shoryuken.options[:groups].to_a.each do |group, options|
-        Shoryuken.add_group(group, options.fetch(:concurrency, 25))
+        Shoryuken.add_group(group, options.fetch(:concurrency, 25), delay: options.fetch(:delay, 0))
         options[:queues].to_a.each do |queue, weight|
           parse_queue(queue, weight, group)
         end

lib/shoryuken/launcher.rb

@@ -72,7 +72,7 @@ def create_managers
       Shoryuken.groups.map do |group, options|
         Shoryuken::Manager.new(
           Shoryuken::Fetcher.new(group),
-          Shoryuken.polling_strategy(group).new(options[:queues]),
+          Shoryuken.polling_strategy(group).new(options[:queues], Shoryuken.options.delay(group)),
           options[:concurrency],
           executor
         )

lib/shoryuken/options.rb

@@ -30,9 +30,10 @@ def active_job?
         defined?(::ActiveJob)
       end
 
-      def add_group(group, concurrency)
+      def add_group(group, concurrency, delay: 0)
         groups[group] ||= {
           concurrency: concurrency,
+          delay: delay,
           queues: []
         }
       end
@@ -77,7 +78,6 @@ def launcher_executor=(launcher_executor)
 
       def polling_strategy(group)
         strategy = (group == 'default' ? options : options[:groups].to_h[group]).to_h[:polling_strategy]
-
         case strategy
         when 'WeightedRoundRobin', nil # Default case
           Polling::WeightedRoundRobin
@@ -90,6 +90,10 @@ def polling_strategy(group)
         end
       end
 
+      def delay(group)
+        groups[group].to_h.fetch(:delay, options[:delay]).to_f
+      end
+
       def start_callback
         @@start_callback
       end

lib/shoryuken/polling/base.rb

@@ -57,10 +57,8 @@ def ==(other)
         end
       end
 
-      private
-
       def delay
-        Shoryuken.options[:delay].to_f
+        @delay || Shoryuken.options[:delay].to_f
       end
     end
   end

lib/shoryuken/polling/strict_priority.rb

@@ -1,7 +1,7 @@
 module Shoryuken
   module Polling
     class StrictPriority < BaseStrategy
-      def initialize(queues)
+      def initialize(queues, delay = nil)
         # Priority ordering of the queues, highest priority first
         @queues = queues
                   .group_by { |q| q }
@@ -12,6 +12,7 @@ def initialize(queues)
         @paused_until = queues
                         .each_with_object({}) { |queue, h| h[queue] = Time.at(0) }
 
+        @delay = delay
         # Start queues at 0
         reset_next_queue
       end

lib/shoryuken/polling/weighted_round_robin.rb

@@ -1,10 +1,11 @@
 module Shoryuken
   module Polling
     class WeightedRoundRobin < BaseStrategy
-      def initialize(queues)
+      def initialize(queues, delay = nil)
         @initial_queues = queues
         @queues = queues.dup.uniq
         @paused_queues = []
+        @delay = delay
       end
 
       def next_queue

spec/shoryuken/environment_loader_spec.rb

@@ -4,7 +4,7 @@
 RSpec.describe Shoryuken::EnvironmentLoader do
   subject { described_class.new({}) }
 
-  describe '#parse_queues' do
+  describe '#parse_queues loads default queues' do
     before do
       allow(subject).to receive(:load_rails)
       allow(subject).to receive(:prefix_active_job_queue_names)
@@ -22,6 +22,29 @@
     end
   end
 
+  describe '#parse_queues includes delay per groups' do
+    before do
+      allow(subject).to receive(:load_rails)
+      allow(subject).to receive(:prefix_active_job_queue_names)
+      allow(subject).to receive(:require_workers)
+      allow(subject).to receive(:validate_queues)
+      allow(subject).to receive(:validate_workers)
+      allow(subject).to receive(:patch_deprecated_workers)
+    end
+
+    specify do
+      Shoryuken.options[:queues] = ['queue1', 'queue2'] # default queues
+      Shoryuken.options[:groups] = [[ 'custom', { queues: ['queue3'], delay: 25 }]]
+      subject.load
+
+      expect(Shoryuken.groups['default'][:queues]).to eq(%w[queue1 queue2])
+      expect(Shoryuken.groups['default'][:delay]).to eq(0)
+      expect(Shoryuken.groups['custom'][:queues]).to eq(%w[queue3])
+      expect(Shoryuken.groups['custom'][:delay]).to eq(25)
+    end
+  end
+
+
   describe '#prefix_active_job_queue_names' do
     before do
       allow(subject).to receive(:load_rails)

spec/shoryuken/options_spec.rb

@@ -1,19 +1,35 @@
 require 'spec_helper'
 
 RSpec.describe Shoryuken::Options do
-  describe '.add_group' do
+  describe '.add_group adds queues and optional delay' do
     before do
       Shoryuken.groups.clear
       Shoryuken.add_group('group1', 25)
       Shoryuken.add_group('group2', 25)
+      Shoryuken.add_group('group3', 25, delay: 5)
     end
 
     specify do
       described_class.add_queue('queue1', 1, 'group1')
       described_class.add_queue('queue2', 2, 'group2')
+      described_class.add_queue('queue3', 1, 'group3')
 
       expect(described_class.groups['group1'][:queues]).to eq(%w[queue1])
       expect(described_class.groups['group2'][:queues]).to eq(%w[queue2 queue2])
+      expect(described_class.groups['group3'][:queues]).to eq(%w[queue3])
+      expect(described_class.groups['group3'][:delay]).to eq(5)
+    end
+  end
+
+  describe '.delay works for each group' do
+    specify do
+      Shoryuken.add_group('group1', 25)
+      Shoryuken.add_group('group2', 25, delay: 5)
+      described_class.add_queue('queue1', 1, 'group1')
+      described_class.add_queue('queue2', 2, 'group2')
+
+      expect(described_class.delay('group1')).to eq(0.0)
+      expect(described_class.delay('group2')).to eq(5.0)
     end
   end
 

spec/shoryuken/polling/strict_priority_spec.rb

@@ -86,6 +86,14 @@
     end
   end
 
+  describe 'delay' do
+    it 'sets delay based on group' do
+      delay_polling = Shoryuken::Polling::StrictPriority.new(queues, 25)
+      expect(delay_polling.delay).to eq(25.0)
+      expect(subject.delay).to eq(1.0)
+    end
+  end
+
   describe '#messages_found' do
     it 'pauses a queue if there are no messages found' do
       # [shoryuken, 2]

spec/shoryuken/polling/weighted_round_robin_spec.rb

@@ -96,4 +96,12 @@
       expect(subject.instance_variable_get(:@queues)).to eq([queue1, queue2, queue1])
     end
   end
+
+  describe 'delay' do
+    it 'sets delay based on group' do
+      delay_polling = Shoryuken::Polling::WeightedRoundRobin.new(queues, 25)
+      expect(delay_polling.delay).to eq(25.0)
+      expect(subject.delay).to eq(1.0)
+    end
+  end
 end