Class: AceMQ::AMQP::Topology

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/topology.rb

Overview

The exchanges, queues and bindings a service expects to find.

Declaring them one call at a time works, and stops working the moment somebody needs to know what a service will do to a broker before it does it. A topology can be printed and read in a deployment review, checked for the mistakes that are cheaper to catch here than on the broker, and then applied — which is the difference between a change somebody approved and one they found out about.

topology = Topology.new
.exchange("orders-events", :topic)
.queue("shipping-orders", dead_letter: true)
.binding("shipping-orders", "orders-events", "order.placed")
.binding("shipping-orders", "orders-events", "order.cancelled")

puts topology            # what it would do
topology.apply(connection)

Every builder method returns self and mutates in place, which is the one place in this library that is not a frozen value object: a topology is assembled and then applied, and threading a new copy through six chained calls would buy immutability nobody is using.

Defined Under Namespace

Classes: Binding, Exchange, Queue

Constant Summary collapse

DEAD_LETTER_EXCHANGE =

The exchange dead letters reach their queue through.

The name itself lives in Naming, with the retry exchange it is a pair with, so that the two strings the broker's shape depends on are written down in one place and read everywhere else.

Naming::DEAD_LETTER_EXCHANGE
RETRY_EXCHANGE =

The exchange a retry rung expires through.

Unlike DEAD_LETTER_EXCHANGE this one cannot be renamed per topology, and that is deliberate rather than an omission. The dead-letter exchange is only ever named by the queue arguments this topology writes itself, so a team on a shared vhost can point it somewhere inside their own prefix without anybody else noticing. The retry exchange is written into the rung's argument table, which is the table five libraries have to agree on: a topology that renamed it would declare rungs that answer PRECONDITION_FAILED to the very consumer meant to publish into them.

Naming::RETRY_EXCHANGE

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(dead_letter_exchange: DEAD_LETTER_EXCHANGE) ⇒ Topology

Returns a new instance of Topology.

Parameters:

  • dead_letter_exchange (String) (defaults to: DEAD_LETTER_EXCHANGE) —

    where dead letters are routed through. The shared name by default, which is what makes a broker legible; nameable because a shared vhost with one team per prefix has no business declaring an exchange outside its own.



82
83
84
85
86
87
 
# File 'lib/acemq/amqp/topology.rb', line 82

def initialize(dead_letter_exchange: DEAD_LETTER_EXCHANGE)
  @dead_letter_exchange = dead_letter_exchange
  @exchanges = []
  @queues = []
  @bindings = []
end

Instance Attribute Details

#bindings ⇒ Object (readonly)

Returns the value of attribute bindings.



76
77
78
 
# File 'lib/acemq/amqp/topology.rb', line 76

def bindings
  @bindings
end

#dead_letter_exchange ⇒ Object (readonly)

Returns the value of attribute dead_letter_exchange.



76
77
78
 
# File 'lib/acemq/amqp/topology.rb', line 76

def dead_letter_exchange
  @dead_letter_exchange
end

#exchanges ⇒ Object (readonly)

Returns the value of attribute exchanges.



76
77
78
 
# File 'lib/acemq/amqp/topology.rb', line 76

def exchanges
  @exchanges
end

#queues ⇒ Object (readonly)

Returns the value of attribute queues.



76
77
78
 
# File 'lib/acemq/amqp/topology.rb', line 76

def queues
  @queues
end

Instance Method Details

#apply(connection) ⇒ Topology

Declares everything, in the order a broker needs: exchanges, then queues, then the bindings between them.

It stops at the first failure. A queue that already exists with different settings is refused with PRECONDITION_FAILED, and that refusal is passed on rather than swallowed: it means this service and the broker disagree about what the queue is, and carrying on would leave the disagreement in place with nobody told.

Parameters:

  • connection (Connection, Transport) —

    anything answering declare_exchange, declare_queue and bind

Returns:



312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
 
# File 'lib/acemq/amqp/topology.rb', line 312

def apply(connection)
  validate!
  @exchanges.each do |e|
    connection.declare_exchange(
      e.name, kind: e.kind, durable: e.durable,
              auto_delete: e.auto_delete, arguments: e.arguments
    )
  end
  # The kind goes out as well as the arguments, even though the arguments
  # already carry it. A connection defaults an unqualified declaration to
  # quorum, and a plan that said classic and then let the default answer
  # for it would declare something nobody reviewed.
  @queues.each do |q|
    connection.declare_queue(q.name, queue_type: q.type, durable: q.durable,
                                     auto_delete: q.auto_delete,
                                     exclusive: q.exclusive, arguments: q.arguments)
  end
  @bindings.each do |b|
    connection.bind(queue: b.queue, exchange: b.exchange, routing_key: b.routing_key)
  end
  self
end

#binding(queue, exchange, routing_key = "") ⇒ Topology

Routes messages matching a key from an exchange to a queue.

Returns:



208
209
210
211
212
 
# File 'lib/acemq/amqp/topology.rb', line 208

def binding(queue, exchange, routing_key = "")
  @bindings << Binding.new(queue: queue.to_s, exchange: exchange.to_s,
                           routing_key: routing_key.to_s)
  self
end

#dead_letter_queue(source) ⇒ Topology

Adds the dead-letter queue for a source queue, and the exchange that reaches it.

Called for you by queue(..., dead_letter: true). It is public because a service that only consumes dead letters needs the queue declared without declaring the source queue it drains.

Classic, like the rungs and for the same reason: Java declares {source}.dlq CLASSIC, and a dead-letter queue is a queue two services in two languages both declare before either of them drains it.

Parameters:

  • source (String) —

    the queue whose dead letters these are

Returns:



227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/acemq/amqp/topology.rb', line 227

def dead_letter_queue(source)
  source = source.to_s
  dlq = Naming.dead_letter_queue(source)
  unless declared_exchange?(@dead_letter_exchange)
    exchange(@dead_letter_exchange,
             :direct)
  end
  add_queue(dlq, QueueType::CLASSIC, durable: true, auto_delete: false,
                                     exclusive: false, arguments: {})
  binding(dlq, @dead_letter_exchange, dlq)
end

#exchange(name, kind, durable: true, auto_delete: false, arguments: {}) ⇒ Topology

Adds a durable exchange.

Parameters:

  • name (String)
  • kind (Symbol, String) —

    direct, topic, fanout or headers

Returns:



94
95
96
97
98
 
# File 'lib/acemq/amqp/topology.rb', line 94

def exchange(name, kind, durable: true, auto_delete: false, arguments: {})
  @exchanges << Exchange.new(name: name.to_s, kind: kind.to_s, durable: durable,
                             auto_delete: auto_delete, arguments: arguments)
  self
end

#parked_queue(source) ⇒ Topology

Adds the parking queue for a source queue, where messages that cannot even be decoded go.

Separate from the dead-letter queue on purpose: a message that failed five times and a message nothing could read are two different problems, and mixing them means whoever drains the dead letters has to sort them by hand.

Classic, for the reason #dead_letter_queue is.

Parameters:

  • source (String)

Returns:



251
252
253
254
255
256
257
258
259
260
 
# File 'lib/acemq/amqp/topology.rb', line 251

def parked_queue(source)
  parked = Naming.parked_queue(source.to_s)
  unless declared_exchange?(@dead_letter_exchange)
    exchange(@dead_letter_exchange,
             :direct)
  end
  add_queue(parked, QueueType::CLASSIC, durable: true, auto_delete: false,
                                        exclusive: false, arguments: {})
  binding(parked, @dead_letter_exchange, parked)
end

#plan ⇒ Array<String>

What applying it would do, in the order a broker needs it.

Deliberately not a diff against the live broker: AMQP offers no way to enumerate what is there without the management API, and a plan that quietly guessed would be worse than one honest about being a statement of intent.

Returns:

  • (Array<String>) 


293
294
295
296
297
298
 
# File 'lib/acemq/amqp/topology.rb', line 293

def plan
  validate!
  @exchanges.map { |e| "declare exchange #{e.name} (#{describe_exchange(e)})" } +
    @queues.map { |q| "declare queue #{q.name} (#{describe_queue(q)})" } +
    @bindings.map { |b| "bind #{b.queue} to #{b.exchange} on #{b.routing_key.inspect}" }
end

#problems ⇒ Array<String>

What is wrong with the description itself, before any of it reaches a broker.

A binding naming a queue nothing declares is the mistake worth catching here. The broker would accept it if the queue happened to exist already, and the service would then quietly depend on something no deployment creates — which works until the day it is deployed somewhere new.

Returns:

  • (Array<String>) —

    every problem found, empty when there are none



272
273
274
 
# File 'lib/acemq/amqp/topology.rb', line 272

def problems
  blank_names + duplicates + missing_kinds + unresolved_bindings
end

#queue(name, durable: true, auto_delete: false, exclusive: false, arguments: {}, queue_type: nil, dead_letter: false, retry_policy: nil, retry_threshold: RetryLadder::DEFAULT_THRESHOLD) ⇒ Topology

Adds a durable quorum queue, which is the default everywhere in this library.

The kind matters as much as the name. RabbitMQ treats x-queue-type as part of a queue's identity, so a queue that exists as a quorum queue answers PRECONDITION_FAILED to anybody declaring it classic — and that anybody is the second service to start, in whichever language it happens to be written. Java declares a source queue quorum, Java is the one with deployments, so the other four libraries declare it quorum too and a Ruby service can share orders with a Java one.

queue_type: :classic still gets a classic queue for a caller who wants one, and a x-queue-type in arguments is honoured as it always was, which is how Patterns.declare_stream declares a stream. A queue that is exclusive, auto-deleting or transient is classic whatever the default says, because RabbitMQ refuses to replicate a queue that goes away on its own; asking for quorum and one of those flags is a QueueTypeError rather than a silent downgrade.

dead_letter wires the queue to its own dead-letter queue, using the names from Naming so that the dead letters of orders.new are in orders.new.dlq whichever language declared them. It adds the queue, the shared exchange and the binding between them, so a caller does not have to remember three declarations to get one behaviour right.

retry_policy adds the rungs that policy needs: the queues a retry waits in when its delay is long enough that waiting in the consumer would lose it to a restart. A policy is the right thing to hand a topology because the rungs are RetryPolicy#schedule — a finite list of delays, known before anything is published — so they can be declared once, here, and appear in a plan somebody reviews, rather than being discovered one failure at a time. Passing the policy rather than a list of delays is what keeps the two from drifting: a queue whose consumer runs a policy this was never told about has rungs that do not match its waits, and the symptom of that is a retry which quietly never comes back.

Parameters:

  • name (String)
  • queue_type (Symbol, nil) (defaults to: nil) —

    :quorum by default, :classic or :stream for a caller who needs one

  • dead_letter (Boolean) (defaults to: false) —

    whether to wire up {name}.dlq

  • retry_policy (RetryPolicy, nil) (defaults to: nil) —

    whose rungs to declare

  • retry_threshold (Numeric) (defaults to: RetryLadder::DEFAULT_THRESHOLD) —

    seconds; delays at or above it get a rung, and must match what the consumer of this queue is configured with

Returns:

Raises:

  • (QueueTypeError) —

    when the kind asked for and the flags asked for cannot both be had



148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/acemq/amqp/topology.rb', line 148

def queue(name, durable: true, auto_delete: false, exclusive: false, arguments: {},
          queue_type: nil, dead_letter: false, retry_policy: nil,
          retry_threshold: RetryLadder::DEFAULT_THRESHOLD)
  name = name.to_s
  arguments = dead_letter_arguments(name, arguments) if dead_letter
  add_queue(name, queue_type, durable: durable, auto_delete: auto_delete,
                              exclusive: exclusive, arguments: arguments)
  dead_letter_queue(name) if dead_letter
  retry_ladder(name, retry_policy, threshold: retry_threshold) if retry_policy
  self
end

#retry_ladder(source, policy, threshold: RetryLadder::DEFAULT_THRESHOLD) ⇒ Topology

Adds the retry rungs a policy needs for a queue, and the exchange that brings their expired messages home.

Called for you by queue(..., retry_policy: policy). Public for the same reason #dead_letter_queue is: a deployment that only creates the broker's shape, and never consumes, still has to create these.

Nothing binds a consumer to a rung, and nothing should: a rung is only ever published into, and a consumer on one would take the message before its time-to-live had expired, which is the entire wait.

The binding that does get added goes the other way: source is bound to RETRY_EXCHANGE under its own name, which is what carries an expired message home. It is added here rather than left to whoever declares the source queue because a rung without it fails silently — the message enters the rung, the time-to-live runs out, and the broker drops it, having nowhere to route it and nobody to tell. Which is also why source has to be a queue this topology declares: #problems says so, and the alternative is a plan that binds something no deployment creates.

Parameters:

  • source (String) —

    the queue whose retries these are

  • policy (RetryPolicy)
  • threshold (Numeric) (defaults to: RetryLadder::DEFAULT_THRESHOLD) —

    seconds

Returns:



185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/acemq/amqp/topology.rb', line 185

def retry_ladder(source, policy, threshold: RetryLadder::DEFAULT_THRESHOLD)
  source = source.to_s
  ladder = RetryLadder.for(source, policy, threshold: threshold)
  return self if ladder.empty?

  exchange(RETRY_EXCHANGE, :direct) unless declared_exchange?(RETRY_EXCHANGE)
  # Classic, and not because nobody got round to changing it. Java's
  # RetryTopology declares every rung CLASSIC, and a rung is a queue two
  # libraries publish into by name, so the argument table — the queue type
  # included — has to be the same one in both. It is also the queue this
  # design leans on hardest: a rung exists to hold a message until a
  # time-to-live expires it into an exchange, which is the plainest thing
  # a classic queue does.
  ladder.rungs.each do |rung|
    add_queue(rung.queue, QueueType::CLASSIC, durable: true, auto_delete: false,
                                              exclusive: false, arguments: rung.arguments)
  end
  binding(source, RETRY_EXCHANGE, source)
end

#to_s ⇒ Object

The plan, as something worth putting in a deployment log.



336
337
338
339
340
341
342
343
 
# File 'lib/acemq/amqp/topology.rb', line 336

def to_s
  header = "Topology: #{@exchanges.size} exchanges, #{@queues.size} queues, " \
           "#{@bindings.size} bindings"
  found = problems
  return "#{header}\n  invalid: #{found.join("\n  invalid: ")}" unless found.empty?

  ([header] + plan.map { |line| "  #{line}" }).join("\n")
end

#validate! ⇒ Topology

Returns self.

Returns:

Raises:



278
279
280
281
282
283
 
# File 'lib/acemq/amqp/topology.rb', line 278

def validate!
  found = problems
  return self if found.empty?

  raise TopologyError, "this topology cannot be applied:\n  #{found.join("\n  ")}"
end