Class: Racecar::Config

Inherits:
KingKonf::Config
  • Object
show all
Defined in:
lib/racecar/config.rb

Constant Summary collapse

STATISTICS_DISABLED_VALUE = 
0

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(env: ENV) ⇒ Config

Returns a new instance of Config.



214
215
216
217
218
219
 
# File 'lib/racecar/config.rb', line 214

def initialize(env: ENV)
  super(env: env)
  @error_handler = proc {}
  @subscriptions = []
  @logger = Logger.new(STDOUT)
end

Instance Attribute Details

#consumer_class ⇒ Object

Returns the value of attribute consumer_class.



260
261
262
 
# File 'lib/racecar/config.rb', line 260

def consumer_class
  @consumer_class
end

#error_handler ⇒ Object (readonly)

The error handler must be set directly on the object.



198
199
200
 
# File 'lib/racecar/config.rb', line 198

def error_handler
  @error_handler
end

#instrumenter ⇒ Object 



282
283
284
285
286
287
288
289
290
291
292
293
294
295
 
# File 'lib/racecar/config.rb', line 282

def instrumenter
  @instrumenter ||= begin
    default_payload = { client_id: client_id, group_id: group_id }

    if defined?(ActiveSupport::Notifications)
      # ActiveSupport needs `concurrent-ruby` but doesn't `require` it.
      require 'concurrent/utility/monotonic_time'
      Instrumenter.new(backend: ActiveSupport::Notifications, default_payload: default_payload)
    else
      logger.warn "ActiveSupport::Notifications not available, instrumentation is disabled"
      NullInstrumenter
    end
  end
end

#logger ⇒ Object

Returns the value of attribute logger.



200
201
202
 
# File 'lib/racecar/config.rb', line 200

def logger
  @logger
end

#parallel_workers ⇒ Object

Returns the value of attribute parallel_workers.



200
201
202
 
# File 'lib/racecar/config.rb', line 200

def parallel_workers
  @parallel_workers
end

#subscriptions ⇒ Object

Returns the value of attribute subscriptions.



200
201
202
 
# File 'lib/racecar/config.rb', line 200

def subscriptions
  @subscriptions
end

Instance Method Details

#inspect ⇒ Object 



221
222
223
224
225
226
 
# File 'lib/racecar/config.rb', line 221

def inspect
  self.class.variables
    .map(&:name)
    .map {|key| [key, get(key).inspect].join(" = ") }
    .join("\n")
end

#install_liveness_probe ⇒ Object 



298
299
300
 
# File 'lib/racecar/config.rb', line 298

def install_liveness_probe
  liveness_probe.tap(&:install)
end

#liveness_probe ⇒ Object 



302
303
304
305
306
307
308
309
 
# File 'lib/racecar/config.rb', line 302

def liveness_probe
  require "active_support/notifications"
  @liveness_probe ||= LivenessProbe.new(
    ActiveSupport::Notifications,
    liveness_probe_file_path,
    liveness_probe_max_interval
  )
end

#load_consumer_class(consumer_class) ⇒ Object 



242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
 
# File 'lib/racecar/config.rb', line 242

def load_consumer_class(consumer_class)
  self.consumer_class = consumer_class
  self.group_id = consumer_class.group_id || self.group_id

  self.group_id ||= [
    # Configurable and optional prefix:
    group_id_prefix,

    # MyFunnyConsumer => my-funny-consumer
    consumer_class.name.gsub(/[a-z][A-Z]/) { |str| "#{str[0]}-#{str[1]}" }.downcase,
  ].compact.join

  self.parallel_workers = consumer_class.parallel_workers
  self.subscriptions = consumer_class.subscriptions
  self.max_wait_time = consumer_class.max_wait_time || self.max_wait_time
  self.fetch_messages = consumer_class.fetch_messages || self.fetch_messages
  self.pidfile ||= "#{group_id}.pid"
end

#max_wait_time_ms ⇒ Object 



210
211
212
 
# File 'lib/racecar/config.rb', line 210

def max_wait_time_ms
  max_wait_time * 1000
end

#on_error(&handler) ⇒ Object 



262
263
264
 
# File 'lib/racecar/config.rb', line 262

def on_error(&handler)
  @error_handler = handler
end

#rdkafka_consumer ⇒ Object 



266
267
268
269
270
271
272
 
# File 'lib/racecar/config.rb', line 266

def rdkafka_consumer
  consumer_config = consumer.map do |param|
    param.split("=", 2).map(&:strip)
  end.to_h
  consumer_config.merge!(rdkafka_security_config)
  consumer_config
end

#rdkafka_producer ⇒ Object 



274
275
276
277
278
279
280
 
# File 'lib/racecar/config.rb', line 274

def rdkafka_producer
  producer_config = producer.map do |param|
    param.split("=", 2).map(&:strip)
  end.to_h
  producer_config.merge!(rdkafka_security_config)
  producer_config
end

#statistics_interval_ms ⇒ Object 



202
203
204
205
206
207
208
 
# File 'lib/racecar/config.rb', line 202

def statistics_interval_ms
  if Rdkafka::Config.statistics_callback
    statistics_interval * 1000
  else
    STATISTICS_DISABLED_VALUE
  end
end

#validate! ⇒ Object 



228
229
230
231
232
233
234
235
236
237
238
239
240
 
# File 'lib/racecar/config.rb', line 228

def validate!
  if brokers.empty?
    raise ConfigError, "`brokers` must not be empty"
  end

  if socket_timeout <= max_wait_time
    raise ConfigError, "`socket_timeout` must be longer than `max_wait_time`"
  end

  if max_pause_timeout && !pause_with_exponential_backoff?
    raise ConfigError, "`max_pause_timeout` only makes sense when `pause_with_exponential_backoff` is enabled"
  end
end