mirror of
https://github.com/mperham/sidekiq.git
synced 2022-11-09 13:52:34 -05:00
a2dd36c6d1
At the moment, `MyWorker.perform_in(1.month)` always schedules a job in 30 days. When added to a date, `1.month` will add 1 to the date's month count, which means that it will add 28, 29, 30, or 31 days depending on the month and year. When I call `MyWorker.perform_in(1.month)`, I would expect the job to be scheduled next month, same day of the month, all the time. At the moment, it is true only four months in the year. My pull request tries to fix this by converting the interval object to a Float at the last possible moment. Plaese note that the test I wrote will fail only during months that do not have 30 days. Ideally, I would add a dependency to Timecop and freeze time to any day in a month of 28, 29 or 31 days. This could also avoid using `#assert_in_delta`, in favour of `#assert_equal`. Feel free to read my blog post [Rails' `1.month` has a variable length](http://dstosik.github.io/rails/2015/02/19/rails-1month-variable-length/) for more details.
89 lines
2.5 KiB
Ruby
89 lines
2.5 KiB
Ruby
require 'sidekiq/client'
|
|
require 'sidekiq/core_ext'
|
|
|
|
module Sidekiq
|
|
|
|
##
|
|
# Include this module in your worker class and you can easily create
|
|
# asynchronous jobs:
|
|
#
|
|
# class HardWorker
|
|
# include Sidekiq::Worker
|
|
#
|
|
# def perform(*args)
|
|
# # do some work
|
|
# end
|
|
# end
|
|
#
|
|
# Then in your Rails app, you can do this:
|
|
#
|
|
# HardWorker.perform_async(1, 2, 3)
|
|
#
|
|
# Note that perform_async is a class method, perform is an instance method.
|
|
module Worker
|
|
attr_accessor :jid
|
|
|
|
def self.included(base)
|
|
base.extend(ClassMethods)
|
|
base.class_attribute :sidekiq_options_hash
|
|
base.class_attribute :sidekiq_retry_in_block
|
|
base.class_attribute :sidekiq_retries_exhausted_block
|
|
end
|
|
|
|
def logger
|
|
Sidekiq.logger
|
|
end
|
|
|
|
module ClassMethods
|
|
|
|
def perform_async(*args)
|
|
client_push('class' => self, 'args' => args)
|
|
end
|
|
|
|
def perform_in(interval, *args)
|
|
int = interval.to_f
|
|
now = Time.now
|
|
ts = (int < 1_000_000_000 ? (now + interval).to_f : int)
|
|
|
|
item = { 'class' => self, 'args' => args, 'at' => ts }
|
|
|
|
# Optimization to enqueue something now that is scheduled to go out now or in the past
|
|
item.delete('at'.freeze) if ts <= now.to_f
|
|
|
|
client_push(item)
|
|
end
|
|
alias_method :perform_at, :perform_in
|
|
|
|
##
|
|
# Allows customization for this type of Worker.
|
|
# Legal options:
|
|
#
|
|
# :queue - use a named queue for this Worker, default 'default'
|
|
# :retry - enable the RetryJobs middleware for this Worker, default *true*
|
|
# :backtrace - whether to save any error backtrace in the retry payload to display in web UI,
|
|
# can be true, false or an integer number of lines to save, default *false*
|
|
# :pool - use the given Redis connection pool to push this type of job to a given shard.
|
|
def sidekiq_options(opts={})
|
|
self.sidekiq_options_hash = get_sidekiq_options.merge((opts || {}).stringify_keys)
|
|
end
|
|
|
|
def sidekiq_retry_in(&block)
|
|
self.sidekiq_retry_in_block = block
|
|
end
|
|
|
|
def sidekiq_retries_exhausted(&block)
|
|
self.sidekiq_retries_exhausted_block = block
|
|
end
|
|
|
|
def get_sidekiq_options # :nodoc:
|
|
self.sidekiq_options_hash ||= Sidekiq.default_worker_options
|
|
end
|
|
|
|
def client_push(item) # :nodoc:
|
|
pool = Thread.current[:sidekiq_via_pool] || get_sidekiq_options['pool'] || Sidekiq.redis_pool
|
|
Sidekiq::Client.new(pool).push(item.stringify_keys)
|
|
end
|
|
|
|
end
|
|
end
|
|
end
|