mirror of
https://github.com/mperham/sidekiq.git
synced 2022-11-09 13:52:34 -05:00
74 lines
2.5 KiB
Ruby
74 lines
2.5 KiB
Ruby
require 'sidekiq/middleware/chain'
|
|
require 'sidekiq/middleware/client/unique_jobs'
|
|
|
|
module Sidekiq
|
|
class Client
|
|
|
|
def self.default_middleware
|
|
Middleware::Chain.new do |m|
|
|
end
|
|
end
|
|
|
|
def self.registered_workers
|
|
Sidekiq.redis { |x| x.smembers('workers') }
|
|
end
|
|
|
|
def self.registered_queues
|
|
Sidekiq.redis { |x| x.smembers('queues') }
|
|
end
|
|
|
|
##
|
|
# The main method used to push a job to Redis. Accepts a number of options:
|
|
#
|
|
# queue - the named queue to use, default 'default'
|
|
# class - the worker class to call, required
|
|
# args - an array of simple arguments to the perform method, must be JSON-serializable
|
|
# retry - whether to retry this job if it fails, true or false, default true
|
|
# backtrace - whether to save any error backtrace, default false
|
|
#
|
|
# All options must be strings, not symbols. NB: because we are serializing to JSON, all
|
|
# symbols in 'args' will be converted to strings.
|
|
#
|
|
# Example:
|
|
# Sidekiq::Client.push('queue' => 'my_queue', 'class' => MyWorker, 'args' => ['foo', 1, :bat => 'bar'])
|
|
#
|
|
def self.push(item)
|
|
raise(ArgumentError, "Message must be a Hash of the form: { 'class' => SomeWorker, 'args' => ['bob', 1, :foo => 'bar'] }") unless item.is_a?(Hash)
|
|
raise(ArgumentError, "Message must include a class and set of arguments: #{item.inspect}") if !item['class'] || !item['args']
|
|
raise(ArgumentError, "Message must include a Sidekiq::Worker class, not class name: #{item['class'].ancestors.inspect}") if !item['class'].is_a?(Class) || !item['class'].respond_to?('get_sidekiq_options')
|
|
|
|
worker_class = item['class']
|
|
item['class'] = item['class'].to_s
|
|
|
|
item = worker_class.get_sidekiq_options.merge(item)
|
|
item['retry'] = !!item['retry']
|
|
queue = item['queue']
|
|
|
|
pushed = false
|
|
Sidekiq.client_middleware.invoke(worker_class, item, queue) do
|
|
payload = Sidekiq.dump_json(item)
|
|
Sidekiq.redis do |conn|
|
|
if item['at']
|
|
pushed = (conn.zadd('schedule', item['at'].to_s, payload) == 1)
|
|
else
|
|
_, pushed = conn.multi do
|
|
conn.sadd('queues', queue)
|
|
conn.rpush("queue:#{queue}", payload)
|
|
end
|
|
end
|
|
end
|
|
end
|
|
!! pushed
|
|
end
|
|
|
|
# Redis compatibility helper. Example usage:
|
|
#
|
|
# Sidekiq::Client.enqueue(MyWorker, 'foo', 1, :bat => 'bar')
|
|
#
|
|
# Messages are enqueued to the 'default' queue.
|
|
#
|
|
def self.enqueue(klass, *args)
|
|
klass.perform_async(*args)
|
|
end
|
|
end
|
|
end
|