Class: AceMQ::Rails::Configuration

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/rails/configuration.rb

Overview

Everything this integration can be told, in one object.

Two ways in, and they are not alternatives:

  • config/acemq.yml, read through Rails' own config_for, which already understands per-environment sections, a shared: block and ERB. That is where a URL, a virtual host and a topology belong: they differ per environment and they are data.
  • config.acemq.* in config/application.rb or an environment file, for the things that are decisions rather than data — a codec object, an interceptor, a telemetry reporter. None of those can be written in YAML without inventing a second configuration language.

The YAML is read first and the Ruby is applied over it. A file checked in for every environment is the general statement; a line in config/environments/test.rb is the specific one, and the specific one wins. This is the opposite order from database.yml, which has no Ruby half to disagree with, and the same order as config.active_job.queue_adapter overriding whatever an adapter's own defaults were.

Defined Under Namespace

Classes: ConsumerDefaults

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Configuration

Returns a new instance of Configuration.



180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
 
# File 'lib/acemq/rails/configuration.rb', line 180

def initialize
  @url = "amqp://guest:guest@localhost:5672"
  @connect_timeout = 10
  @heartbeat = :server
  @publisher_confirms = true
  @format = "json"
  @declare_topology_on_boot = true
  @shutdown_timeout = 20
  @connect_on_boot = false
  @transport_options = {}
  @topology = {}
  @interceptors = []
  @consumer_paths = []
  @consumer = ConsumerDefaults.new
end

Instance Attribute Details

#client_name ⇒ Object

What is stamped on every message this application publishes, in x-acemq-origin. Defaults to the Rails application's name and the hostname, which is what makes a dead-lettered message traceable to a pod.



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

def client_name
  @client_name
end

#codec ⇒ Object

A codec instance, for anything that cannot be named — a composite, an encrypted one, a claim-check wrapper.



112
113
114
 
# File 'lib/acemq/rails/configuration.rb', line 112

def codec
  @codec
end

#connect_on_boot ⇒ Object

Whether opening the connection is attempted while the web server boots. False by default, and that default is a decision: a broker that is down should not stop an application from serving the pages that do not need it. The connection is opened on first use instead.



167
168
169
 
# File 'lib/acemq/rails/configuration.rb', line 167

def connect_on_boot
  @connect_on_boot
end

#connect_timeout ⇒ Object

Seconds to wait for the AMQP handshake.



63
64
65
 
# File 'lib/acemq/rails/configuration.rb', line 63

def connect_timeout
  @connect_timeout
end

#consumer ⇒ Object

Consumer defaults, overridable per consumer class.



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

def consumer
  @consumer
end

#consumer_paths ⇒ Object

Extra directories to autoload consumers from.

Empty by default and that is not an omission: Rails already autoloads every direct subdirectory of app/, so a consumer in app/consumers needs nothing here. This is for lib/consumers and the like.



178
179
180
 
# File 'lib/acemq/rails/configuration.rb', line 178

def consumer_paths
  @consumer_paths
end

#declare_topology_on_boot ⇒ Object

Whether to apply topology when the consumer process starts. True by default: a consumer that starts before its queue exists reads nothing and says nothing about why.



153
154
155
 
# File 'lib/acemq/rails/configuration.rb', line 153

def declare_topology_on_boot
  @declare_topology_on_boot
end

#format ⇒ Object

The codec name, resolved through AceMQ::AMQP::Codecs.build: bytes, json, string, toml, xml, yaml. AceMQ::AMQP::Codecs.names is the list, and a name that is not on it raises.

Protobuf and Avro are deliberately not names. Each needs something a string cannot carry — a generated message class, a schema, a registry — so each is built and handed over as an object in codec instead. See docs/serialization.md.



108
109
110
 
# File 'lib/acemq/rails/configuration.rb', line 108

def format
  @format
end

#heartbeat ⇒ Object

Heartbeat interval in seconds, or :server to take the broker's suggestion. Taking the broker's suggestion is right almost always.



67
68
69
 
# File 'lib/acemq/rails/configuration.rb', line 67

def heartbeat
  @heartbeat
end

#interceptors ⇒ Object

Interceptors applied to the process connection as it is opened.

An interceptor lives on a connection — intercept_publish and intercept_consume are instance methods on AMQP::Connection — so registering one from an initializer means reaching for AceMQ::Rails.connection, and that opens the socket during boot. It is the one thing connect_on_boot: false exists to avoid, and it happens quietly: the application dials the broker on the way up and nothing in the configuration says why. This list is applied on the way out of Connection.open instead, so an interceptor costs nothing until something publishes or consumes.

Each entry is registered on both sides. The library works out which hooks an object actually answers to once, at registration, so an object with only before_publish is a publish interceptor and nothing else — which is exactly how the library's own AceMQ::AMQP::Telemetry::OpenTelemetry#install registers itself. A bare object is accepted as well as an array.

Ruby rather than YAML, because an interceptor is an object. Applied only by AceMQ::Rails.connection: a connection handed in with AceMQ::Rails.connection= is a test's own, and is left as it was given.



144
145
146
 
# File 'lib/acemq/rails/configuration.rb', line 144

def interceptors
  @interceptors
end

#max_outstanding_publishes ⇒ Object

How many publishes may be waiting for a confirm at once. This is the back pressure that stops a runaway loop from holding a million messages in this process's memory. 0.6.0 of the library added it; the default is the library's.



73
74
75
 
# File 'lib/acemq/rails/configuration.rb', line 73

def max_outstanding_publishes
  @max_outstanding_publishes
end

#password ⇒ Object

Broker login, kept out of the URL so it stays out of the log line that reports a failed connection. Either both or neither.



47
48
49
 
# File 'lib/acemq/rails/configuration.rb', line 47

def password
  @password
end

#publisher_confirms ⇒ Object

Whether a publish waits for the broker to confirm it. True, and the writer refuses anything else.

Confirms are not optional in the library and there is no keyword to ask for a publish without them: the publishing channel calls confirm_select the moment it is opened, and publish raises AMQP::PublishError when nothing comes back. The setting stays because it has been documented as one since 0.1.0, and it refuses rather than being ignored — a line in a configuration file that reads as though it turned durability off, and did nothing at all, is worse than no line.



85
86
87
 
# File 'lib/acemq/rails/configuration.rb', line 85

def publisher_confirms
  @publisher_confirms
end

#shutdown_timeout ⇒ Object

Seconds the consumer process spends draining before it gives up. Must be comfortably shorter than whatever will kill the process — terminationGracePeriodSeconds, TimeoutStopSec, docker stop -t.



161
162
163
 
# File 'lib/acemq/rails/configuration.rb', line 161

def shutdown_timeout
  @shutdown_timeout
end

#telemetry ⇒ Object

A telemetry reporter: anything answering count, observe and gauge. Nil means the library's own no-op, which costs nothing.



120
121
122
 
# File 'lib/acemq/rails/configuration.rb', line 120

def telemetry
  @telemetry
end

#tls ⇒ Object

An AceMQ::AMQP::Security, for a private certificate authority or a client certificate. Nil means "whatever the URL scheme implies".



116
117
118
 
# File 'lib/acemq/rails/configuration.rb', line 116

def tls
  @tls
end

#token ⇒ Object

A token instead of a username and password, for brokers that take one (RabbitMQ with OAuth 2). Mutually exclusive with username.



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

def token
  @token
end

#topology ⇒ Object

Declared on boot of the consumer process, and by rake acemq:topology. A hash of exchanges, queues and bindings; see TopologyBuilder.



148
149
150
 
# File 'lib/acemq/rails/configuration.rb', line 148

def topology
  @topology
end

#transport_options ⇒ Object

Extra keyword arguments handed to Bunny.new, for the handful of things this configuration does not name.



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

def transport_options
  @transport_options
end

#url ⇒ Object

The AMQP URL. amqps:// is verified against the system trust store on its own; a broker with a private certificate authority needs tls.



43
44
45
 
# File 'lib/acemq/rails/configuration.rb', line 43

def url
  @url
end

#username ⇒ Object

Broker login, kept out of the URL so it stays out of the log line that reports a failed connection. Either both or neither.



47
48
49
 
# File 'lib/acemq/rails/configuration.rb', line 47

def username
  @username
end

#virtual_host ⇒ Object

The AMQP virtual host. Nil leaves whatever is in the URL alone; a value replaces the URL's path, which is how a vhost is expressed in AMQP.



55
56
57
 
# File 'lib/acemq/rails/configuration.rb', line 55

def virtual_host
  @virtual_host
end

Instance Method Details

#assign(settings) ⇒ Configuration

Applies a hash read out of config/acemq.yml.

String keys, because that is what YAML gives, and dashes accepted alongside underscores because max-outstanding-publishes is how the same setting is spelled in the Spring starter's application.yaml and somebody will copy it across.

An unknown key raises rather than being ignored. A typo in a configuration file that is silently dropped is how a production broker ends up with the default retry policy and nobody knows why.

Parameters:

  • settings (Hash)

Returns:



273
274
275
276
277
278
279
280
281
282
283
284
285
286
 
# File 'lib/acemq/rails/configuration.rb', line 273

def assign(settings)
  return self if settings.nil?

  settings.each do |key, value|
    name = normalise(key)
    case name
    when "consumer" then @consumer.assign(symbolize(value))
    when "topology" then @topology = value
    when "transport_options" then @transport_options = symbolize(value)
    else assign_one(name, value)
    end
  end
  self
end

#connection_arguments ⇒ Array(String, Hash)

What AMQP::Connection.open is called with.

Returns:

  • (Array(String, Hash)) 


340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
 
# File 'lib/acemq/rails/configuration.rb', line 340

def connection_arguments
  options = {
    codec: resolved_codec,
    origin: client_name,
    retry_policy: consumer.retry_policy,
    prefetch: consumer.prefetch,
    retry_threshold: consumer.broker_wait_threshold,
    telemetry: telemetry,
    heartbeat: heartbeat,
    connection_timeout: connect_timeout,
    security: tls,
    credentials: resolved_credentials
  }
  # Only when asked for. The library's default is its own business and
  # repeating it here would freeze it at whatever it was the day this was
  # written.
  if max_outstanding_publishes
    options[:max_outstanding_publishes] = max_outstanding_publishes.to_i
  end
  options.compact!
  [resolved_url, options.merge(transport_options)]
end

#resolved_codec ⇒ #encode

The codec this configuration asks for.

Returns:

  • (#encode) 


291
292
293
294
295
 
# File 'lib/acemq/rails/configuration.rb', line 291

def resolved_codec
  return AceMQ::AMQP::Codec.check!(codec) if codec

  AceMQ::AMQP::Codecs.build(format)
end

#resolved_credentials ⇒ AceMQ::AMQP::Credentials?

The credentials this configuration asks for, or nil to leave the URL to speak for itself.

Returns:

  • (AceMQ::AMQP::Credentials, nil) 


301
302
303
304
305
306
307
 
# File 'lib/acemq/rails/configuration.rb', line 301

def resolved_credentials
  return AceMQ::AMQP::Credentials.token(token) if token

  return nil if username.nil? || username.to_s.empty?

  AceMQ::AMQP::Credentials.of(username: username, password: password.to_s)
end

#resolved_url ⇒ String

The URL with virtual_host applied.

A vhost is the path of an AMQP URL, and this is fiddlier than joining two strings because "/" is a legal vhost name — it is RabbitMQ's default — and also the separator. The value here is taken literally, as the name of the vhost, and escaped whole: / becomes /%2F and tenant/a becomes /tenant%2Fa, which is the vhost really called "tenant/a" rather than a path with two segments.

CGI.escape with its +++ put back, because CGI.escape encodes a space the way a form does and a URI path wants %20. A vhost with a space in it is rare and is not a reason for this to be wrong.

Returns:

  • (String) 


323
324
325
326
327
328
329
330
331
332
333
334
335
 
# File 'lib/acemq/rails/configuration.rb', line 323

def resolved_url
  return url.to_s if virtual_host.nil? || virtual_host.to_s.empty?

  require "uri"
  require "cgi"
  parsed = URI.parse(url.to_s)
  parsed.path = "/#{CGI.escape(virtual_host.to_s).gsub("+", "%20")}"
  parsed.to_s
rescue URI::InvalidURIError => e
  raise AceMQ::AMQP::ConfigurationError,
        "acemq: the url #{AceMQ::AMQP::Transport.redact(url.to_s)} cannot be " \
        "parsed, so virtual_host cannot be applied to it: #{e.message}"
end