Class: Shoryuken::Options

Inherits:
Object
  • Object
show all
Defined in:
lib/shoryuken/options.rb

Overview

Stores and manages all Shoryuken configuration options. This class is used internally to hold settings for workers, queues, middleware, and other runtime configurations.

Constant Summary collapse

DEFAULTS =

Default configuration values for Shoryuken

{
  thread_priority: -1,
  concurrency: 25,
  queues: [],
  aws: {},
  delay: 0.0,
  timeout: 8,
  lifecycle_events: {
    startup: [],
    dispatch: [],
    utilization_update: [],
    quiet: [],
    shutdown: [],
    stopped: []
  }
}.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Options

Initializes a new Options instance with default values



52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/shoryuken/options.rb', line 52

def initialize
  self.groups = {}
  self.worker_registry = DefaultWorkerRegistry.new
  self.exception_handlers = [DefaultExceptionHandler]
  self.active_job_queue_name_prefixing = false
  self.active_job_fifo_message_deduplication = true
  self.worker_executor = Worker::DefaultExecutor
  self.cache_visibility_timeout = false
  self.reloader = proc { |&block| block.call }
  self.enable_reloading ||= false
  # this is needed for keeping backward compatibility
  @sqs_client_receive_message_opts ||= {}
end

Instance Attribute Details

#active_job_fifo_message_deduplication ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def active_job_fifo_message_deduplication
  @active_job_fifo_message_deduplication
end

#active_job_queue_name_prefixing ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def active_job_queue_name_prefixing
  @active_job_queue_name_prefixing
end

#cache_visibility_timeout ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def cache_visibility_timeout
  @cache_visibility_timeout
end

#default_worker_options ⇒ Hash{String => Object}

Returns the default worker options hash

Returns:

  • (Hash{String => Object}) —

    the default worker options



237
238
239
240
241
242
243
244
245
246
 
# File 'lib/shoryuken/options.rb', line 237

def default_worker_options
  @default_worker_options ||= {
    'queue' => 'default',
    'delete' => false,
    'auto_delete' => false,
    'auto_visibility_timeout' => false,
    'retry_intervals' => nil,
    'batch' => false
  }
end

#enable_reloading ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def enable_reloading
  @enable_reloading
end

#exception_handlers ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def exception_handlers
  @exception_handlers
end

#groups ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def groups
  @groups
end

#launcher_executor ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def launcher_executor
  @launcher_executor
end

#logger ⇒ Logger

Returns the logger instance

Returns:

  • (Logger) —

    the logger



173
174
175
 
# File 'lib/shoryuken/options.rb', line 173

def logger
  @logger ||= Shoryuken::Logging.logger
end

#reloader ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def reloader
  @reloader
end

#sqs_client ⇒ Aws::SQS::Client

Returns the SQS client, initializing a default one if needed. Uses AWS configuration from options if available.

Returns:

  • (Aws::SQS::Client) —

    the SQS client



151
152
153
 
# File 'lib/shoryuken/options.rb', line 151

def sqs_client
  @sqs_client ||= Aws::SQS::Client.new(options[:aws])
end

#sqs_client_receive_message_opts ⇒ Hash

Returns options passed to SQS receive_message calls.

Returns:

  • (Hash) —

    options passed to SQS receive_message calls



49
50
51
 
# File 'lib/shoryuken/options.rb', line 49

def sqs_client_receive_message_opts
  @sqs_client_receive_message_opts
end

#start_callback ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def start_callback
  @start_callback
end

#stop_callback ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def stop_callback
  @stop_callback
end

#thread_priority ⇒ Integer

Returns the thread priority setting

Returns:

  • (Integer) —

    the thread priority



180
181
182
 
# File 'lib/shoryuken/options.rb', line 180

def thread_priority
  @thread_priority ||= options[:thread_priority]
end

#worker_executor ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def worker_executor
  @worker_executor
end

#worker_registry ⇒ Boolean, ...

Returns:

  • (Boolean) —

    whether to enable ActiveJob queue name prefixing

  • (Boolean) —

    whether to cache SQS visibility timeout

  • (Hash{String => Hash}) —

    the configured processing groups

  • (Object) —

    the executor used to launch workers

  • (Proc) —

    the code reloader proc for development environments

  • (Boolean) —

    whether code reloading is enabled

  • (Proc, nil) —

    callback to execute when server starts

  • (Proc, nil) —

    callback to execute when server stops

  • (Class) —

    the executor class for running workers

  • (Shoryuken::WorkerRegistry) —

    the registry for worker classes

  • (Array<#call>) —

    handlers for processing exceptions



37
38
39
 
# File 'lib/shoryuken/options.rb', line 37

def worker_registry
  @worker_registry
end

Instance Method Details

#active_job? ⇒ Boolean

Checks if ActiveJob is available

Returns:

  • (Boolean) —

    true if ActiveJob is defined



69
70
71
 
# File 'lib/shoryuken/options.rb', line 69

def active_job?
  defined?(::ActiveJob)
end

#active_job_fifo_message_deduplication? ⇒ Boolean

Checks if the ActiveJob adapter should auto-generate a content-based message_deduplication_id for FIFO queues. When disabled, distinct enqueues of the same job class and arguments are no longer silently deduplicated.

Returns:

  • (Boolean) —

    true if FIFO deduplication id generation is enabled



312
313
314
 
# File 'lib/shoryuken/options.rb', line 312

def active_job_fifo_message_deduplication?
  @active_job_fifo_message_deduplication
end

#active_job_queue_name_prefixing? ⇒ Boolean

Checks if ActiveJob queue name prefixing is enabled

Returns:

  • (Boolean) —

    true if prefixing is enabled



303
304
305
 
# File 'lib/shoryuken/options.rb', line 303

def active_job_queue_name_prefixing?
  @active_job_queue_name_prefixing
end

#add_group(group, concurrency = nil, delay: nil, polling_strategy: nil) ⇒ Hash

Adds a processing group with the specified concurrency and delay

Parameters:

  • group (String) —

    the name of the group

  • concurrency (Integer, nil) (defaults to: nil) —

    the number of concurrent workers for the group

  • delay (Float, nil) (defaults to: nil) —

    the delay between polling cycles

  • polling_strategy (Class, String, nil) (defaults to: nil) —

    the polling strategy class for the group

Returns:

  • (Hash) —

    the group configuration

Raises:



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/shoryuken/options.rb', line 81

def add_group(group, concurrency = nil, delay: nil, polling_strategy: nil)
  unless polling_strategy.nil? || polling_strategy.is_a?(Class) || polling_strategy.is_a?(String)
    raise Errors::InvalidPollingStrategyError, "#{polling_strategy} is not a valid polling_strategy"
  end

  concurrency ||= options[:concurrency]
  delay ||= options[:delay]

  groups[group] ||= {
    concurrency: concurrency,
    delay: delay,
    polling_strategy: polling_strategy,
    queues: []
  }
end

#add_queue(queue, weight, group) ⇒ void

This method returns an undefined value.

Adds a queue to a processing group with the specified weight

Parameters:

  • queue (String) —

    the name of the queue

  • weight (Integer) —

    the weight (priority) of the queue

  • group (String) —

    the name of the group to add the queue to



103
104
105
106
107
 
# File 'lib/shoryuken/options.rb', line 103

def add_queue(queue, weight, group)
  weight.times do
    groups[group][:queues] << queue
  end
end

#cache_visibility_timeout? ⇒ Boolean

Checks if visibility timeout caching is enabled

Returns:

  • (Boolean) —

    true if caching is enabled



296
297
298
 
# File 'lib/shoryuken/options.rb', line 296

def cache_visibility_timeout?
  @cache_visibility_timeout
end

#client_middleware {|Shoryuken::Middleware::Chain| ... } ⇒ Shoryuken::Middleware::Chain

Returns the client middleware chain

Yields:

Returns:



228
229
230
231
232
 
# File 'lib/shoryuken/options.rb', line 228

def client_middleware
  @_client_chain ||= default_client_middleware
  yield @_client_chain if block_given?
  @_client_chain
end

#configure_client {|Shoryuken::Options| ... } ⇒ void

This method returns an undefined value.

Yields self unless running as a server for client-specific configuration

Yields:



220
221
222
 
# File 'lib/shoryuken/options.rb', line 220

def configure_client
  yield self unless server?
end

#configure_server {|Shoryuken::Options| ... } ⇒ void

This method returns an undefined value.

Yields self if running as a server for server-specific configuration

Yields:



202
203
204
 
# File 'lib/shoryuken/options.rb', line 202

def configure_server
  yield self if server?
end

#delay(group) ⇒ Float

Returns the polling delay for a group

Parameters:

  • group (String) —

    the name of the group

Returns:

  • (Float) —

    the delay in seconds



143
144
145
 
# File 'lib/shoryuken/options.rb', line 143

def delay(group)
  groups[group].to_h.fetch(:delay, options[:delay]).to_f
end

#on(event, &block) { ... } ⇒ void

This method returns an undefined value.

Registers a block to run at a point in the Shoryuken lifecycle.

Examples:

Shoryuken.configure_server do |config|
  config.on(:shutdown) do
    puts "Goodbye cruel world!"
  end
end

Parameters:

  • event (Symbol) —

    the lifecycle event (:startup, :quiet, :shutdown, or :stopped)

  • block (Proc) —

    the block to execute for the event

Yields:

  • the block to execute for the event

Raises:

  • (ArgumentError) —

    if event is not a Symbol or not a valid event name



279
280
281
282
283
284
 
# File 'lib/shoryuken/options.rb', line 279

def on(event, &block)
  raise Errors::InvalidEventError, "Symbols only please: #{event}" unless event.is_a?(Symbol)
  raise Errors::InvalidEventError, "Invalid event name: #{event}" unless options[:lifecycle_events].key?(event)

  options[:lifecycle_events][event] << block
end

#on_start(&block) { ... } ⇒ void

This method returns an undefined value.

Registers a callback to run when the server starts

Parameters:

  • block (Proc) —

    the block to execute on start

Yields:

  • the block to execute on start



253
254
255
 
# File 'lib/shoryuken/options.rb', line 253

def on_start(&block)
  self.start_callback = block
end

#on_stop(&block) { ... } ⇒ void

This method returns an undefined value.

Registers a callback to run when the server stops

Parameters:

  • block (Proc) —

    the block to execute on stop

Yields:

  • the block to execute on stop



262
263
264
 
# File 'lib/shoryuken/options.rb', line 262

def on_stop(&block)
  self.stop_callback = block
end

#options ⇒ Hash

Returns the global options hash

Returns:

  • (Hash) —

    the options hash



166
167
168
 
# File 'lib/shoryuken/options.rb', line 166

def options
  @options ||= DEFAULTS.dup
end

#polling_strategy(group) ⇒ Class

Returns the polling strategy class for a group

Parameters:

  • group (String) —

    the name of the group

Returns:

  • (Class) —

    the polling strategy class to use



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/shoryuken/options.rb', line 120

def polling_strategy(group)
  strategy = groups[group].to_h[:polling_strategy] ||
             (group == 'default' ? options : options[:groups].to_h[group]).to_h[:polling_strategy]
  case strategy
  when 'WeightedRoundRobin', nil # Default case
    Polling::WeightedRoundRobin
  when 'StrictPriority'
    Polling::StrictPriority
  when String
    begin
      Object.const_get(strategy)
    rescue NameError
      raise Errors::InvalidPollingStrategyError, "#{strategy} is not a valid polling_strategy"
    end
  when Class
    strategy
  end
end

#register_worker(*args) ⇒ void

This method returns an undefined value.

Registers a worker class with the worker registry

Parameters:

  • args (Array) —

    arguments to pass to the registry



194
195
196
 
# File 'lib/shoryuken/options.rb', line 194

def register_worker(*args)
  worker_registry.register_worker(*args)
end

#server? ⇒ Boolean

Checks if running as a server (CLI mode)

Returns:

  • (Boolean) —

    true if Shoryuken::CLI is defined



289
290
291
 
# File 'lib/shoryuken/options.rb', line 289

def server?
  defined?(Shoryuken::CLI)
end

#server_middleware {|Shoryuken::Middleware::Chain| ... } ⇒ Shoryuken::Middleware::Chain

Returns the server middleware chain

Yields:

Returns:



210
211
212
213
214
 
# File 'lib/shoryuken/options.rb', line 210

def server_middleware
  @_server_chain ||= default_server_middleware
  yield @_server_chain if block_given?
  @_server_chain
end

#ungrouped_queues ⇒ Array<String>

Returns all queues from all groups

Returns:

  • (Array<String>) —

    flat array of all queue names



112
113
114
 
# File 'lib/shoryuken/options.rb', line 112

def ungrouped_queues
  groups.values.flat_map { |options| options[:queues] }
end