Merge pull request #28 from cavi21/perform_in_ability

Add the ability to schedule sidekiq jobs

.gitignore

@@ -23,4 +23,5 @@ tmp
 mkmf.log
 .ruby-version
 .ruby-gemset
-.idea
+.idea
+.gs

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## [Unreleased]
 
+- Add the ability to pass `sidekiq_schedule_options` options in order to schedule jobs to be run in the future
+
 ## [1.2.0]
 
 - fixes: gemspec does not allow sidekiq 5.x to be used

Gemfile

@@ -7,6 +7,8 @@ gem 'rake'
 gem 'rspec'
 gem 'coveralls', require: false
 
+gem 'redis', '<= 4.0.3' if RUBY_VERSION < '2.3'
+
 gem 'psych', platforms: :rbx
 
 group :extras do

README.md

@@ -51,6 +51,40 @@ for more information.
 
 In order to define custom [sidekiq_options](https://github.com/mperham/sidekiq/wiki/Advanced-Options#workers) you can add `sidekiq_options` class method in your subscriber definition - those options will be passed to Sidekiq's `set` method just before scheduling the asynchronous worker.
 
+### Passing down schedule options
+
+In order be able to schedule jobs to be run in the future following [Scheduled Jobs](https://github.com/mperham/sidekiq/wiki/Scheduled-Jobs) you can add `sidekiq_schedule_options` class method in your subscriber definition - those options will be passed to Sidekiq's `perform_in` method when the worker is called.
+
+This feature is not as powerfull as Sidekiq's API that allows you to set this on every job enqueue, in this case you're able to set this for the hole listener class like:
+```ruby
+class MyListener
+  #...
+
+  def self.sidekiq_schedule_options
+    { perform_in: 5 }
+  end
+
+  #...
+end
+```
+Or you can set this per event (method called on the listener), like so:
+```ruby
+class MyListener
+  #...
+
+  def self.sidekiq_schedule_options
+    { event_name: { perform_in: 5 } }
+  end
+
+  def self.event_name
+    #...
+  end
+
+  #...
+end
+```
+In both cases there is also available the `perform_at` option.
+
 ## Compatibility
 
 The same Ruby versions as Sidekiq are offically supported, but it should work

Rakefile

@@ -1,2 +1 @@
 require "bundler/gem_tasks"
-

lib/wisper/sidekiq.rb

@@ -28,14 +28,37 @@ def self.register
 
     def broadcast(subscriber, publisher, event, args)
       options = sidekiq_options(subscriber)
-      Worker.set(options).perform_async(::YAML.dump([subscriber, event, args]))
+      schedule_options = sidekiq_schedule_options(subscriber, event)
+
+      Worker.set(options).perform_in(
+        schedule_options.fetch(:delay, 0),
+        ::YAML.dump([subscriber, event, args])
+      )
     end
 
     private
 
     def sidekiq_options(subscriber)
       subscriber.respond_to?(:sidekiq_options) ? subscriber.sidekiq_options : {}
     end
+
+    def sidekiq_schedule_options(subscriber, event)
+      return {} unless subscriber.respond_to?(:sidekiq_schedule_options)
+
+      options = subscriber.sidekiq_schedule_options
+
+      if options.has_key?(event.to_sym)
+        delay_option(options[event.to_sym])
+      else
+        delay_option(options)
+      end
+    end
+
+    def delay_option(options)
+      return {} unless options.key?(:perform_in) || options.key?(:perform_at)
+
+      { delay: options[:perform_in] || options[:perform_at] }
+    end
   end
 end
 

spec/wisper/sidekiq_broadcaster_spec.rb

@@ -23,6 +23,60 @@ def self.sidekiq_options
     end
   end
 
+  class CustomizedScheduleInJobSubscriberUnderTest
+    def self.it_happened
+    end
+
+    def self.sidekiq_schedule_options
+      { perform_in: 5 }
+    end
+  end
+
+  class CustomizedEventScheduleInJobSubscriberUnderTest
+    def self.it_happened
+    end
+
+    def self.sidekiq_schedule_options
+      { it_happened: { perform_in: 5 } }
+    end
+  end
+
+  class CustomizedScheduleAtJobSubscriberUnderTest
+    def self.it_happened
+    end
+
+    def self.sidekiq_schedule_options
+      { perform_at: Time.now + 5 }
+    end
+  end
+
+  class CustomizedEventScheduleAtJobSubscriberUnderTest
+    def self.it_happened
+    end
+
+    def self.sidekiq_schedule_options
+      { it_happened: { perform_at: Time.now + 5 } }
+    end
+  end
+
+  class CustomizedBadScheduleInJobSubscriberUnderTest
+    def self.it_happened
+    end
+
+    def self.sidekiq_schedule_options
+      { perform_in: 'not a number', delay: 5 }
+    end
+  end
+
+  class CustomizedBadDefaultScheduleInWithEventScheduleAtJobSubscriberUnderTest
+    def self.it_happened
+    end
+
+    def self.sidekiq_schedule_options
+      { perform_in: 'not a number', delay: 5, it_happened: { perform_at: Time.now + 5 } }
+    end
+  end
+
   let(:publisher) { PublisherUnderTest.new }
 
   before { Sidekiq::Testing.fake! }
@@ -36,13 +90,61 @@ def self.sidekiq_options
         .to change(Sidekiq::Queues["default"], :size).by(1)
     end
 
+    it 'schedules to run in some time a sidekiq job' do
+      publisher.subscribe(CustomizedScheduleInJobSubscriberUnderTest, async: described_class.new)
+
+      # In order to look into Sidekiq::ScheduledSet we need to hit redis
+      expect { publisher.run }
+        .to change { Sidekiq::Queues["default"].select{|job| job.key?('at')}.size }.by(1)
+    end
+
+    it 'schedules to run in some time a sidekiq job for an event' do
+      publisher.subscribe(CustomizedEventScheduleInJobSubscriberUnderTest, async: described_class.new)
+
+      # In order to look into Sidekiq::ScheduledSet we need to hit redis
+      expect { publisher.run }
+        .to change { Sidekiq::Queues["default"].select{|job| job.key?('at')}.size }.by(1)
+    end
+
+    it 'schedules to run at some time a sidekiq job' do
+      publisher.subscribe(CustomizedEventScheduleAtJobSubscriberUnderTest, async: described_class.new)
+
+      # In order to look into Sidekiq::ScheduledSet we need to hit redis
+      expect { publisher.run }
+        .to change { Sidekiq::Queues["default"].select{|job| job.key?('at')}.size }.by(1)
+    end
+
+    it 'schedules to run at some time a sidekiq job for an event' do
+      publisher.subscribe(CustomizedEventScheduleInJobSubscriberUnderTest, async: described_class.new)
+
+      # In order to look into Sidekiq::ScheduledSet we need to hit redis
+      expect { publisher.run }
+        .to change { Sidekiq::Queues["default"].select{|job| job.key?('at')}.size }.by(1)
+    end
+
     it 'can respect custom sidekiq_options' do
       publisher.subscribe(CustomizedSubscriberUnderTest, async: described_class.new)
 
       expect { publisher.run }
         .to change(Sidekiq::Queues["my_queue"], :size).by(1)
     end
 
+    it 'schedules a sidekiq job with bad sidekiq_schedule_options' do
+      publisher.subscribe(CustomizedBadScheduleInJobSubscriberUnderTest, async: described_class.new)
+
+      expect { publisher.run }
+        .to change(Sidekiq::Queues["default"], :size).by(1)
+      expect { publisher.run }
+        .not_to change { Sidekiq::Queues["default"].select{|job| job.key?('at')}.size }
+    end
+
+    it 'schedules a sidekiq job with bad sidekiq_schedule_options' do
+      publisher.subscribe(CustomizedBadDefaultScheduleInWithEventScheduleAtJobSubscriberUnderTest, async: described_class.new)
+
+      expect { publisher.run }
+        .to change { Sidekiq::Queues["default"].select{|job| job.key?('at')}.size }.by(1)
+    end
+
     context 'when provides subscriber with args' do
       let(:subscriber) { RegularSubscriberUnderTest }
       let(:event) { 'it_happened' }