class Roundhouse::Client

Attributes

redis_pool[RW]

Public Class Methods

default() click to toggle source

deprecated

# File lib/roundhouse/client.rb, line 130
def default
  @default ||= new
end
enqueue_to(queue_id, klass, *args) click to toggle source

Example usage:

Roundhouse::Client.enqueue_to(queue_id, MyWorker, 'foo', 1, :bat => 'bar')
# File lib/roundhouse/client.rb, line 148
def enqueue_to(queue_id, klass, *args)
  klass.client_push('queue_id' => queue_id, 'class' => klass, 'args' => args)
end
enqueue_to_in(queue_id, interval, klass, *args) click to toggle source

Example usage:

Roundhouse::Client.enqueue_to_in(queue_id, 3.minutes, MyWorker, 'foo', 1, :bat => 'bar')
# File lib/roundhouse/client.rb, line 155
def enqueue_to_in(queue_id, interval, klass, *args)
  int = interval.to_f
  now = Time.now.to_f
  ts = (int < 1_000_000_000 ? now + int : int)

  item = { 'class' => klass, 'args' => args, 'at' => ts, 'queue_id' => queue_id }
  item.delete('at'.freeze) if ts <= now

  klass.client_push(item)
end
new(redis_pool=nil) click to toggle source

Roundhouse::Client normally uses the default Redis pool but you may pass a custom ConnectionPool if you want to shard your Roundhouse jobs across several Redis instances (for scalability reasons, e.g.)

Roundhouse::Client.new(ConnectionPool.new { Redis.new })

Generally this is only needed for very large Roundhouse installs processing more than thousands jobs per second. I do not recommend sharding unless you truly cannot scale any other way (e.g. splitting your app into smaller apps). Some features, like the API, do not support sharding: they are designed to work against a single Redis instance only.

# File lib/roundhouse/client.rb, line 43
def initialize(redis_pool=nil)
  @redis_pool = redis_pool || Thread.current[:roundhouse_via_pool] || Roundhouse.redis_pool
end
push(item) click to toggle source
# File lib/roundhouse/client.rb, line 134
def push(item)
  new.push(item)
end
push_bulk(items) click to toggle source
# File lib/roundhouse/client.rb, line 138
def push_bulk(items)
  new.push_bulk(items)
end
via(pool) { || ... } click to toggle source

Allows sharding of jobs across any number of Redis instances. All jobs defined within the block will use the given Redis connection pool.

pool = ConnectionPool.new { Redis.new }
Roundhouse::Client.via(pool) do
  SomeWorker.perform_async(1,2,3)
  SomeOtherWorker.perform_async(1,2,3)
end

Generally this is only needed for very large Roundhouse installs processing more than thousands jobs per second. I do not recommend sharding unless you truly cannot scale any other way (e.g. splitting your app into smaller apps). Some features, like the API, do not support sharding: they are designed to work against a single Redis instance.

# File lib/roundhouse/client.rb, line 116
def self.via(pool)
  raise NotImplementedError, 'Roundhouse does not support sharding at this point.'

  raise ArgumentError, "No pool given" if pool.nil?
  raise RuntimeError, "Roundhouse::Client.via is not re-entrant" if x = Thread.current[:roundhouse_via_pool] && x != pool
  Thread.current[:roundhouse_via_pool] = pool
  yield
ensure
  Thread.current[:roundhouse_via_pool] = nil
end

Public Instance Methods

middleware() { |chain| ... } click to toggle source

Define client-side middleware:

client = Roundhouse::Client.new
client.middleware do |chain|
  chain.use MyClientMiddleware
end
client.push('class' => 'SomeWorker', 'args' => [1,2,3])

All client instances default to the globally-defined Roundhouse.client_middleware but you can change as necessary.

# File lib/roundhouse/client.rb, line 20
def middleware(&block)
  @chain ||= Roundhouse.client_middleware
  if block_given?
    @chain = @chain.dup
    yield @chain
  end
  @chain
end
push(item) click to toggle source

The main method used to push a job to Redis. Accepts a number of options:

queue_id - integer queue_id (required, no 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.

Returns a unique Job ID. If middleware stops the job, nil will be returned instead.

Example:

push('queue' => 1, 'class' => MyWorker, 'args' => ['foo', 1, :bat => 'bar'])
# File lib/roundhouse/client.rb, line 64
def push(item)
  normed = normalize_item(item)
  payload = process_single(item['class'], normed)

  if payload
    raw_push([payload])
    payload['jid']
  end
end
push_bulk(items) click to toggle source

Push a large number of jobs to Redis. In practice this method is only useful if you are pushing tens of thousands of jobs or more, or if you need to ensure that a batch doesn’t complete prematurely. This method basically cuts down on the redis round trip latency.

Note: Roundhouse implementation does not use MULTI, so this is not going to be as fast as Sidekiq. As such, this is not officially supported.

Takes the same arguments as push except that args is expected to be an Array of Arrays. All other keys are duplicated for each job. Each job is run through the client middleware pipeline and each job gets its own Job ID as normal.

Returns an array of the of pushed jobs’ jids. The number of jobs pushed can be less than the number given if the middleware stopped processing for one or more jobs.

# File lib/roundhouse/client.rb, line 90
def push_bulk(items)
  Roundhouse.logger.warn '#push_bulk is not officially supported. Use at your own risk.'
  normed = normalize_item(items)
  payloads = items['args'].map do |args|
    raise ArgumentError, "Bulk arguments must be an Array of Arrays: [[1], [2]]" if !args.is_a?(Array)
    process_single(items['class'], normed.merge('args' => args, 'jid' => SecureRandom.hex(12), 'enqueued_at' => Time.now.to_f))
  end.compact

  raw_push(payloads) if !payloads.empty?
  payloads.collect { |payload| payload['jid'] }
end
raw_push_real(payloads)
Alias for: raw_push

Private Instance Methods

normalize_item(item) click to toggle source
# File lib/roundhouse/client.rb, line 185
def normalize_item(item)
  raise(ArgumentError, "Job must be a Hash with 'class' and 'args' keys: { 'class' => SomeWorker, 'args' => ['bob', 1, :foo => 'bar'] }") unless item.is_a?(Hash) && item.has_key?('class'.freeze) && item.has_key?('args'.freeze)
  raise(ArgumentError, "Queue ID must be an integer") unless item['queue_id'.freeze].is_a?(Fixnum)
  raise(ArgumentError, "Job args must be an Array") unless item['args'].is_a?(Array)
  raise(ArgumentError, "Job class must be either a Class or String representation of the class name") unless item['class'.freeze].is_a?(Class) || item['class'.freeze].is_a?(String)

  normalized_hash(item['class'.freeze])
    .each{ |key, value| item[key] = value if item[key].nil? }

  item['class'.freeze] = item['class'.freeze].to_s
  item['queue_id'.freeze] = item['queue_id'.freeze].to_i
  item['jid'.freeze] ||= SecureRandom.hex(12)
  item['created_at'.freeze] ||= Time.now.to_f
  item
end
normalized_hash(item_class) click to toggle source
# File lib/roundhouse/client.rb, line 201
def normalized_hash(item_class)
  if item_class.is_a?(Class)
    raise(ArgumentError, "Message must include a Roundhouse::Worker class, not class name: #{item_class.ancestors.inspect}") if !item_class.respond_to?('get_roundhouse_options'.freeze)
    item_class.get_roundhouse_options
  else
    Roundhouse.default_worker_options
  end
end
process_single(worker_class, item) click to toggle source
# File lib/roundhouse/client.rb, line 177
def process_single(worker_class, item)
  queue_id = item['queue_id']

  middleware.invoke(worker_class, item, queue_id, @redis_pool) do
    item
  end
end
raw_push(payloads) click to toggle source
# File lib/roundhouse/client.rb, line 170
def raw_push(payloads)
  @redis_pool.with do |conn|
    Roundhouse::Monitor.push_job(conn, payloads)
  end
  true
end
Also aliased as: raw_push_real