Module: AceMQ::AMQP::QueueType

Defined in:
lib/acemq/amqp/queue_type.rb

Overview

Which kind of queue to declare, and what that puts in the argument table.

RabbitMQ decides a queue's kind from one argument, x-queue-type, and then treats that argument as part of the queue's identity: a queue that exists as quorum answers PRECONDITION_FAILED to anyone declaring it as anything else. So the kind is not a local preference. It is part of the contract five libraries have to agree on, in the same way the retry rung's arguments are, and it is written down here rather than at each of the half-dozen call sites that declare something.

A durable, named, shared queue is a quorum queue. That is the default in the Java library, which is the one with deployments, and the other four follow it: a replicated queue survives losing the node its leader was on, which is the failure a classic queue turns into lost messages.

The queues this library owns around it stay classic, deliberately: the retry rungs, {queue}.dlq and {queue}.parked. They are declared CLASSIC in Java's RetryTopology and have to be declared the same way here, and a rung's whole behaviour is a time-to-live expiring into a dead-letter exchange, which is what a classic queue does most simply.

Anything exclusive or auto-deleting is classic because it can be nothing else. RabbitMQ rejects an exclusive or auto-delete quorum queue outright, and a quorum queue is always durable. A reply queue belonging to one requester, or a health probe's queue, is therefore left classic rather than being declared into a refusal.

Constant Summary collapse

ARGUMENT =

The argument RabbitMQ reads the kind from.

"x-queue-type"
CLASSIC =

A single-node queue. Fast, and lost with its node.

:classic
QUORUM =

A replicated queue with a consensus protocol behind it.

:quorum
STREAM =

An append-only, replayable log with consumer-held offsets.

:stream
KNOWN = 
[CLASSIC, QUORUM, STREAM].freeze
DEFAULT =

What a durable, named queue is when nobody says otherwise.

QUORUM

Class Method Summary collapse

Class Method Details

.classic_only?(durable: true, exclusive: false, auto_delete: false) ⇒ Boolean

Whether the broker would only ever accept this queue as a classic one.

Quorum queues and streams are replicated on disk, which is why RabbitMQ allows neither to be exclusive, auto-deleting or transient: all three mean "this queue goes away on its own", and there is nothing to replicate about a queue that does.

Returns:

  • (Boolean) 


109
110
111
 
# File 'lib/acemq/amqp/queue_type.rb', line 109

def self.classic_only?(durable: true, exclusive: false, auto_delete: false)
  !durable || exclusive || auto_delete
end

.resolve(name: "", requested: nil, durable: true, exclusive: false, auto_delete: false, arguments: {}) ⇒ Array(Symbol, Hash)

The kind a declaration is asking for, and the arguments that go with it.

type, arguments = QueueType.resolve(name: "orders")
# => [:quorum, { "x-queue-type" => "quorum" }]

The kind can be said three ways and they are read in this order: the requested keyword, an x-queue-type the caller put in the argument table themselves (which is how a stream is declared), and finally the default. Two of them disagreeing is refused rather than resolved, since either answer would be a guess at which one was meant.

A classic queue carries no x-queue-type at all, which is what Java sends and therefore what the broker has to be told: an argument table that differs from another library's is a PRECONDITION_FAILED for whichever service declares second.

Parameters:

  • name (String) (defaults to: "") —

    only ever used to say which queue a refusal is about

  • requested (Symbol, String, nil) (defaults to: nil) —

    :classic, :quorum, :stream

  • durable (Boolean) (defaults to: true)
  • exclusive (Boolean) (defaults to: false)
  • auto_delete (Boolean) (defaults to: false)
  • arguments (Hash) (defaults to: {}) —

    the caller's argument table

Returns:

  • (Array(Symbol, Hash)) —

    the kind, and the arguments to declare with

Raises:

  • (QueueTypeError) —

    when the kind asked for is not one the broker would accept for a queue declared this way



91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/acemq/amqp/queue_type.rb', line 91

def self.resolve(name: "", requested: nil, durable: true, exclusive: false,
                 auto_delete: false, arguments: {})
  requested = known!(name, requested)
  declared = known!(name, arguments[ARGUMENT] || arguments[ARGUMENT.to_sym])
  impossible = classic_only?(durable: durable, exclusive: exclusive,
                             auto_delete: auto_delete)
  type = agreed(name, requested, declared) || (impossible ? CLASSIC : DEFAULT)
  refuse_impossible(name, type, durable: durable, exclusive: exclusive,
                                auto_delete: auto_delete)
  [type, table(type, arguments)]
end

.table(type, arguments) ⇒ Object

The argument table a queue of this kind is declared with.

Classic queues get the caller's table untouched, which matters more than it looks: adding x-queue-type => classic where Java sends nothing would be a difference the broker judges us on.



118
119
120
121
122
 
# File 'lib/acemq/amqp/queue_type.rb', line 118

def self.table(type, arguments)
  return arguments if type == CLASSIC

  arguments.merge(ARGUMENT => type.to_s)
end