Class: Lepus::Configuration

Inherits:
Object
  • Object
show all
Includes:
Web::ConfigExtensions
Defined in:
lib/lepus/configuration.rb

Overview

The class representing the global Lepus configuration.

Constant Summary collapse

DEFAULT_RABBITMQ_URL = 
"amqp://guest:guest@localhost:5672"
DEFAULT_RECOVERY_ATTEMPTS = 
10
DEFAULT_RECOVERY_INTERVAL = 
5.0
DEFAULT_RECOVER_FROM_CONNECTION_CLOSE = 
true
DEFAULT_CONSUMERS_DIRECTORY = 
Pathname.new("app/consumers")
DEFAULT_PROMETHEUS_BUCKETS = 
[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10].freeze

Instance Attribute Summary collapse

Attributes included from Web::ConfigExtensions

#web_show_all_exchanges

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Configuration

Returns a new instance of Configuration.



64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/lepus/configuration.rb', line 64

def initialize
  @connection_name = "Lepus (#{Lepus::VERSION})"
  @rabbitmq_url = ENV.fetch("RABBITMQ_URL", DEFAULT_RABBITMQ_URL) || DEFAULT_RABBITMQ_URL
  @recovery_attempts = DEFAULT_RECOVERY_ATTEMPTS
  @recovery_interval = DEFAULT_RECOVERY_INTERVAL
  @recover_from_connection_close = DEFAULT_RECOVER_FROM_CONNECTION_CLOSE
  @consumers_directory = DEFAULT_CONSUMERS_DIRECTORY
  @process_heartbeat_interval = 60
  @process_alive_threshold = 5 * 60
  @process_registry_backend = :file
  @application_name = nil
  @management_api_url = nil
  @prometheus_buckets = DEFAULT_PROMETHEUS_BUCKETS
end

Instance Attribute Details

#app_executor ⇒ Class

Returns the Rails executor used to wrap asynchronous operations, defaults to the app executor.

Returns:

  • (Class) —

    the Rails executor used to wrap asynchronous operations, defaults to the app executor

See Also:



32
33
34
 
# File 'lib/lepus/configuration.rb', line 32

def app_executor
  @app_executor
end

#application_name ⇒ Object 



51
52
53
 
# File 'lib/lepus/configuration.rb', line 51

def application_name
  @application_name || connection_name
end

#connection_name ⇒ String

Returns the name for the RabbitMQ connection.

Returns:

  • (String) —

    the name for the RabbitMQ connection.



16
17
18
 
# File 'lib/lepus/configuration.rb', line 16

def connection_name
  @connection_name
end

#consumers_directory ⇒ Pathname

Returns the directory where the consumers are stored.

Returns:

  • (Pathname) —

    the directory where the consumers are stored.



28
29
30
 
# File 'lib/lepus/configuration.rb', line 28

def consumers_directory
  @consumers_directory
end

#management_api_url ⇒ String?

Returns the RabbitMQ Management API URL.

Returns:

  • (String, nil) —

    the RabbitMQ Management API URL.



56
57
58
 
# File 'lib/lepus/configuration.rb', line 56

def management_api_url
  @management_api_url
end

#on_thread_error ⇒ Proc

Returns custom lambda/Proc to call when there’s an error within a Lepus thread that takes the exception raised as argument.

Returns:

  • (Proc) —

    custom lambda/Proc to call when there’s an error within a Lepus thread that takes the exception raised as argument



35
36
37
 
# File 'lib/lepus/configuration.rb', line 35

def on_thread_error
  @on_thread_error
end

#process_alive_threshold ⇒ Integer

Returns the threshold in seconds to consider a process alive. Default is 5 minutes.

Returns:

  • (Integer) —

    the threshold in seconds to consider a process alive. Default is 5 minutes.



41
42
43
 
# File 'lib/lepus/configuration.rb', line 41

def process_alive_threshold
  @process_alive_threshold
end

#process_heartbeat_interval ⇒ Integer

Returns the interval in seconds between heartbeats. Default is 60 seconds.

Returns:

  • (Integer) —

    the interval in seconds between heartbeats. Default is 60 seconds.



38
39
40
 
# File 'lib/lepus/configuration.rb', line 38

def process_heartbeat_interval
  @process_heartbeat_interval
end

#process_registry_backend ⇒ Symbol

Returns the process registry backend to use (:file or :rabbitmq). Default is :file.

Returns:

  • (Symbol) —

    the process registry backend to use (:file or :rabbitmq). Default is :file.



44
45
46
 
# File 'lib/lepus/configuration.rb', line 44

def process_registry_backend
  @process_registry_backend
end

#prometheus_buckets ⇒ Array<Numeric>

Returns histogram buckets (in seconds) used by the prometheus_exporter collector for delivery and publish latency.

Returns:

  • (Array<Numeric>) —

    histogram buckets (in seconds) used by the prometheus_exporter collector for delivery and publish latency.



60
61
62
 
# File 'lib/lepus/configuration.rb', line 60

def prometheus_buckets
  @prometheus_buckets
end

#rabbitmq_url ⇒ String

Returns the connection string for RabbitMQ.

Returns:

  • (String) —

    the connection string for RabbitMQ.



13
14
15
 
# File 'lib/lepus/configuration.rb', line 13

def rabbitmq_url
  @rabbitmq_url
end

#recover_from_connection_close ⇒ Boolean

Returns if the recover_from_connection_close value is set for the RabbitMQ connection.

Returns:

  • (Boolean) —

    if the recover_from_connection_close value is set for the RabbitMQ connection.



19
20
21
 
# File 'lib/lepus/configuration.rb', line 19

def recover_from_connection_close
  @recover_from_connection_close
end

#recovery_attempts ⇒ Integer

Returns max number of recovery attempts, nil means forever.

Returns:

  • (Integer) —

    max number of recovery attempts, nil means forever



22
23
24
 
# File 'lib/lepus/configuration.rb', line 22

def recovery_attempts
  @recovery_attempts
end

#recovery_interval ⇒ Integer

Returns the interval in seconds between network recovery attempts.

Returns:

  • (Integer) —

    the interval in seconds between network recovery attempts.



25
26
27
 
# File 'lib/lepus/configuration.rb', line 25

def recovery_interval
  @recovery_interval
end

Instance Method Details

#build_management_api ⇒ Lepus::Web::ManagementAPI

Builds the Management API client based on configuration.

Returns:



92
93
94
 
# File 'lib/lepus/configuration.rb', line 92

def build_management_api
  Web::ManagementAPI.new(base_url: management_api_url)
end

#build_process_registry_backend ⇒ Lepus::ProcessRegistry::Backend

Builds the process registry backend based on configuration.

Returns:



81
82
83
84
85
86
87
88
 
# File 'lib/lepus/configuration.rb', line 81

def build_process_registry_backend
  case process_registry_backend
  when :rabbitmq
    ProcessRegistry::RabbitmqBackend.new
  else
    ProcessRegistry::FileBackend.new
  end
end

#consumer_middleware_chain ⇒ Lepus::Consumers::MiddlewareChain

Returns the global consumer middleware chain.

Returns:



156
157
158
 
# File 'lib/lepus/configuration.rb', line 156

def consumer_middleware_chain
  @consumer_middleware_chain ||= Lepus::Consumers::MiddlewareChain.new
end

#consumer_middlewares {|chain| ... } ⇒ Lepus::Consumers::MiddlewareChain

Configure global consumer middlewares.

Yields:

  • (chain) —

    Block to configure the middleware chain.

Yield Parameters:

Returns:



164
165
166
167
 
# File 'lib/lepus/configuration.rb', line 164

def consumer_middlewares
  yield(consumer_middleware_chain) if block_given?
  consumer_middleware_chain
end

#create_connection(suffix: nil) ⇒ Bunny::Session

Returns the connection to RabbitMQ.

Parameters:

  • suffix (String) (defaults to: nil) —

    the suffix to add to the connection name

Returns:

  • (Bunny::Session) —

    the connection to RabbitMQ



98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/lepus/configuration.rb', line 98

def create_connection(suffix: nil)
  kwargs = connection_config

  if suffix && connection_name
    kwargs[:connection_name] = "#{connection_name} #{suffix}"
  end

  ::Bunny
    .new(rabbitmq_url, **kwargs)
    .tap { |conn| conn.start }
end

#logger=(value) ⇒ void

This method returns an undefined value.

Parameters:

  • value (Logger) —

    the logger to set



171
172
173
 
# File 'lib/lepus/configuration.rb', line 171

def logger=(value)
  Lepus.logger = value
end

#producer(**options) {|producer_config| ... } ⇒ Object

Configure the producer related settings.

Parameters:

  • options (Hash) —

    the options to assign to the producer configuration

Yields:



130
131
132
133
134
 
# File 'lib/lepus/configuration.rb', line 130

def producer(**options)
  producer_config.assign(options) if options.any?
  yield(producer_config) if block_given?
  producer_config
end

#producer_config ⇒ Lepus::Producers::Config

Returns the producer configuration.

Returns:



137
138
139
 
# File 'lib/lepus/configuration.rb', line 137

def producer_config
  @producer_config ||= Lepus::Producers::Config.new
end

#producer_middleware_chain ⇒ Lepus::Producers::MiddlewareChain

Returns the global producer middleware chain.

Returns:



142
143
144
 
# File 'lib/lepus/configuration.rb', line 142

def producer_middleware_chain
  @producer_middleware_chain ||= Lepus::Producers::MiddlewareChain.new
end

#producer_middlewares {|chain| ... } ⇒ Lepus::Producers::MiddlewareChain

Configure global producer middlewares.

Yields:

  • (chain) —

    Block to configure the middleware chain.

Yield Parameters:

Returns:



150
151
152
153
 
# File 'lib/lepus/configuration.rb', line 150

def producer_middlewares
  yield(producer_middleware_chain) if block_given?
  producer_middleware_chain
end

#worker(*names, **options) ⇒ Object

Configure the worker process that will run the consumers.

Parameters:

  • names (Array<Symbol>) —

    the names of the workers to configure

  • options (Hash) —

    the options to assign to the worker configuration



118
119
120
121
122
123
124
125
126
 
# File 'lib/lepus/configuration.rb', line 118

def worker(*names, **options)
  names << Lepus::Consumers::WorkerFactory::DEFAULT_NAME if names.empty?

  names.map(&:to_s).uniq.each do |pid|
    inst = Lepus::Consumers::WorkerFactory[pid]
    inst.assign(options) if options.any?
    yield(inst) if block_given?
  end
end