Module: AceMQ::Rails::TopologyBuilder

Defined in:
lib/acemq/rails/topology_builder.rb

Overview

Turns the topology section of the configuration into a library AMQP::Topology.

topology:
exchanges:
  - name: orders-events
    type: topic
queues:
  - name: orders.new
    dead_letter: true
    retries: { max_attempts: 5, initial_delay: 1, max_delay: 60 }
bindings:
  - queue: orders.new
    exchange: orders-events
    routing_key: order.#

Declared rather than inferred. A queue that appears because some consumer class mentioned it is a queue whose durability, type and dead-lettering nobody decided, and those three are part of a queue's identity to the broker: get them wrong once and the second service to declare it is refused with PRECONDITION_FAILED and cannot consume at all.

Class Method Summary collapse

Class Method Details

.binding(topology, spec) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



91
92
93
94
95
 
# File 'lib/acemq/rails/topology_builder.rb', line 91

def self.binding(topology, spec)
  queue = require_key(spec, :queue, "topology.bindings")
  exchange = require_key(spec, :exchange, "topology.bindings")
  topology.binding(queue, exchange, (spec[:routing_key] || "").to_s)
end

.build(settings) ⇒ AceMQ::AMQP::Topology

Parameters:

  • settings (Hash) —

    the topology section

Returns:

  • (AceMQ::AMQP::Topology)

Raises:

  • (AceMQ::AMQP::TopologyError) —

    when the declaration cannot stand up



46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/acemq/rails/topology_builder.rb', line 46

def self.build(settings)
  settings = normalise(settings)
  topology = AceMQ::AMQP::Topology.new(
    **{ dead_letter_exchange: settings[:dead_letter_exchange] }.compact
  )

  Array(settings[:exchanges]).each { |it| exchange(topology, normalise(it)) }
  Array(settings[:queues]).each { |it| queue(topology, normalise(it)) }
  Array(settings[:bindings]).each { |it| binding(topology, normalise(it)) }

  topology
end

.dead_letter(spec) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

dead_letter: true wires up the conventional {queue}.dlq and the exchange that reaches it. There is deliberately no way to name a different queue: the name is Naming.dead_letter_queue in all five libraries, and a Ruby service that renamed its own would be draining a queue the Java service beside it has never heard of.



129
130
131
 
# File 'lib/acemq/rails/topology_builder.rb', line 129

def self.dead_letter(spec)
  spec[:dead_letter] ? { dead_letter: true } : {}
end

.declared?(settings) ⇒ Boolean

Whether anything at all was declared. An empty section applies nothing, silently and on purpose: an application whose queues are made by a migration tool should not have to configure this away.

Returns:

  • (Boolean) 


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

def self.declared?(settings)
  settings = normalise(settings)
  %i[exchanges queues bindings].any? { |key| !Array(settings[key]).empty? }
end

.exchange(topology, spec) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



68
69
70
71
72
73
74
75
 
# File 'lib/acemq/rails/topology_builder.rb', line 68

def self.exchange(topology, spec)
  name = require_key(spec, :name, "topology.exchanges")
  kind = spec[:type] || spec[:kind] || "topic"
  topology.exchange(name, kind.to_sym,
                    durable: fetch(spec, :durable, true),
                    auto_delete: fetch(spec, :auto_delete, false),
                    arguments: spec[:arguments] || {})
end

.fetch(spec, key, fallback) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



145
146
147
 
# File 'lib/acemq/rails/topology_builder.rb', line 145

def self.fetch(spec, key, fallback)
  spec.key?(key) ? spec[key] : fallback
end

.normalise(value) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

YAML gives string keys and a Rails config_for gives a ActiveSupport::OrderedOptions with symbol ones. Both arrive here.



153
154
155
156
157
 
# File 'lib/acemq/rails/topology_builder.rb', line 153

def self.normalise(value)
  return {} if value.nil?

  value.to_h { |key, inner| [key.to_s.tr("-", "_").to_sym, inner] }
end

.queue(topology, spec) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/acemq/rails/topology_builder.rb', line 78

def self.queue(topology, spec)
  name = require_key(spec, :name, "topology.queues")
  topology.queue(name,
                 durable: fetch(spec, :durable, true),
                 auto_delete: fetch(spec, :auto_delete, false),
                 exclusive: fetch(spec, :exclusive, false),
                 arguments: spec[:arguments] || {},
                 **queue_kind(spec),
                 **dead_letter(spec))
  retries(topology, name, spec)
end

.queue_kind(spec) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



117
118
119
120
 
# File 'lib/acemq/rails/topology_builder.rb', line 117

def self.queue_kind(spec)
  kind = spec[:type] || spec[:queue_type]
  kind.nil? ? {} : { queue_type: kind.to_sym }
end

.require_key(spec, key, where) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



134
135
136
137
138
139
140
141
142
 
# File 'lib/acemq/rails/topology_builder.rb', line 134

def self.require_key(spec, key, where)
  value = spec[key]
  if value.nil? || value.to_s.empty?
    raise AceMQ::AMQP::TopologyError,
          "acemq: every entry in #{where} needs a #{key}; got #{spec.inspect}"
  end

  value.to_s
end

.retries(topology, name, spec) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The rungs of a retry ladder are queues, and they have to exist before a consumer needs them — a consumer that cannot find its rung falls back to waiting in this process, which is slower, holds a prefetch slot, and makes the next deployment's drain sit through the wait. The library counts acemq.retry.rung.missing every time; declaring them here is how that counter stays at nought.



105
106
107
108
109
110
111
112
113
114
 
# File 'lib/acemq/rails/topology_builder.rb', line 105

def self.retries(topology, name, spec)
  settings = spec[:retries] || spec[:retry]
  return unless settings

  settings = normalise(settings)
  threshold = settings.delete(:broker_wait_threshold)
  policy = Configuration::ConsumerDefaults.new.assign(settings).retry_policy
  topology.retry_ladder(name, policy,
                        **{ threshold: threshold }.compact)
end