Class: AceMQ::AMQP::Connection

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

Overview

A connection to a broker.

Everything a service does with AceMQ goes through one of these: it holds the codec, the origin stamped on what it publishes, and the retry policy its consumers use unless they say otherwise. One per process is the usual shape — a connection is a socket and a heartbeat, and opening one per publish is how a service ends up with a thousand of them.

mq = Connection.open("amqp://guest:guest@localhost:5672",
                   origin: "checkout@#{Socket.gethostname}",
                   retry_policy: RetryPolicy.exponential(5, 1, 60))

mq.publish({ "order_id" => "A-1" }, to: "orders.new")

mq.consume("orders.new") do |message|
place(message.payload) ? Ack.accept : Ack.retry("the warehouse said no")
end

Constant Summary collapse

DEFAULT_PREFETCH =

How many unacknowledged messages a consumer holds by default.

20
DRAIN_TIMEOUT =

Seconds #close will spend draining, across every consumer together.

Twenty, because the number that matters is not this one: it is whatever will kill the process if the drain outlasts it. Kubernetes sends SIGTERM, waits terminationGracePeriodSeconds — thirty by default — and then SIGKILLs; systemd has TimeoutStopSec; docker stop waits ten. Twenty inside a thirty-second grace period leaves ten seconds for the web server, for whatever else is shutting down alongside this, and for the process to actually exit. Setting the two equal means the orchestrator wins the race sometimes, and a shutdown that is correct on most deployments is one nobody debugs until it is not.

The same twenty Go's docs/lifecycle.md recommends, for the same reason. Raise it for handlers that genuinely take longer, and raise the grace period with it.

20

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(transport:, codec: JSONCodec.new, origin: nil, retry_policy: RetryPolicy.none, prefetch: DEFAULT_PREFETCH, retry_threshold: RetryLadder::DEFAULT_THRESHOLD, telemetry: nil) ⇒ Connection

Wraps a transport that is already open.

open covers the ordinary case. This is for a transport that is not reached by a URL — a fake in a test, most of all, which is what makes the retry arithmetic below testable without a broker in the room.



221
222
223
224
225
226
227
228
229
230
231
232
233
234
 
# File 'lib/acemq/amqp/connection.rb', line 221

def initialize(transport:, codec: JSONCodec.new, origin: nil,
               retry_policy: RetryPolicy.none, prefetch: DEFAULT_PREFETCH,
               retry_threshold: RetryLadder::DEFAULT_THRESHOLD, telemetry: nil)
  @telemetry = Telemetry::Reporter.for(telemetry)
  @transport = transport
  @codec = Codec.check!(codec)
  @origin = origin.nil? || origin.to_s.empty? ? self.class.default_origin : origin.to_s
  @retry_policy = retry_policy
  @prefetch = prefetch
  @retry_threshold = retry_threshold
  @interceptors = Interceptors.new
  @consumers = []
  @lock = Mutex.new
end

Instance Attribute Details

#codec ⇒ Object (readonly)

Returns the value of attribute codec.



188
189
190
 
# File 'lib/acemq/amqp/connection.rb', line 188

def codec
  @codec
end

#interceptors ⇒ Object (readonly)

Returns the value of attribute interceptors.



188
189
190
 
# File 'lib/acemq/amqp/connection.rb', line 188

def interceptors
  @interceptors
end

#origin ⇒ Object (readonly)

Returns the value of attribute origin.



188
189
190
 
# File 'lib/acemq/amqp/connection.rb', line 188

def origin
  @origin
end

#prefetch ⇒ Object (readonly)

Returns the value of attribute prefetch.



188
189
190
 
# File 'lib/acemq/amqp/connection.rb', line 188

def prefetch
  @prefetch
end

#retry_policy ⇒ Object (readonly)

Returns the value of attribute retry_policy.



188
189
190
 
# File 'lib/acemq/amqp/connection.rb', line 188

def retry_policy
  @retry_policy
end

#retry_threshold ⇒ Object (readonly)

Returns the value of attribute retry_threshold.



188
189
190
 
# File 'lib/acemq/amqp/connection.rb', line 188

def retry_threshold
  @retry_threshold
end

#telemetry ⇒ Object (readonly)

Returns the value of attribute telemetry.



188
189
190
 
# File 'lib/acemq/amqp/connection.rb', line 188

def telemetry
  @telemetry
end

#transport ⇒ Object (readonly)

Returns the value of attribute transport.



188
189
190
 
# File 'lib/acemq/amqp/connection.rb', line 188

def transport
  @transport
end

Class Method Details

.default_origin ⇒ Object

Names the machine but not the service, which is the most an unconfigured library can honestly say. Set origin: to say more: it is the field somebody reads first when a message arrives from a fleet and the question is which of forty pods sent it.



273
274
275
276
277
 
# File 'lib/acemq/amqp/connection.rb', line 273

def self.default_origin
  "acemq@#{Socket.gethostname}"
rescue StandardError
  "acemq@unknown"
end

.open(url, codec: JSONCodec.new, origin: nil, retry_policy: RetryPolicy.none, prefetch: DEFAULT_PREFETCH, telemetry: nil, retry_threshold: RetryLadder::DEFAULT_THRESHOLD, **transport_options) ⇒ Connection

Opens a connection to a broker.

Parameters:

  • url (String) —

    amqp:// or amqps://

  • codec (#encode) (defaults to: JSONCodec.new) —

    the default codec, JSON unless said otherwise

  • origin (String) (defaults to: nil) —

    what to stamp on published messages, conventionally service@host

  • retry_policy (RetryPolicy) (defaults to: RetryPolicy.none) —

    what consumers use by default

  • prefetch (Integer) (defaults to: DEFAULT_PREFETCH) —

    unacknowledged messages per consumer

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

    seconds; a retry delayed this long or longer waits in the broker rather than in the consumer through

  • transport_options (Hash) —

    passed to Transport.open, which is where security: and credentials: go: an amqps:// URL is verified against the system trust store on its own, and a broker with its own certificate authority, or a password that must not be in the URL, is described by a Security and a AceMQ::AMQP::Credentials

Returns:



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

def self.open(url, codec: JSONCodec.new, origin: nil, retry_policy: RetryPolicy.none,
              prefetch: DEFAULT_PREFETCH, telemetry: nil,
              retry_threshold: RetryLadder::DEFAULT_THRESHOLD, **transport_options)
  new(transport: Transport.open(url, **transport_options), codec: codec, origin: origin,
      retry_policy: retry_policy, prefetch: prefetch, retry_threshold: retry_threshold,
      telemetry: telemetry)
end

Instance Method Details

#apply(topology) ⇒ Topology

Declares everything in a topology.

Parameters:

Returns:



465
 
# File 'lib/acemq/amqp/connection.rb', line 465

def apply(topology) = topology.apply(self)

#bind(**options) ⇒ Object 



532
 
# File 'lib/acemq/amqp/connection.rb', line 532

def bind(**options) = @transport.bind(**options)

#blocked? ⇒ Boolean

Whether the broker has asked this connection to stop publishing.

Returns:

  • (Boolean) 


497
 
# File 'lib/acemq/amqp/connection.rb', line 497

def blocked? = !blocked_reason.nil?

#blocked_reason ⇒ String?

Why the broker has asked this connection to stop publishing, or nil.

RabbitMQ blocks a connection when it is running low on memory or disk, and stops reading the socket. A publish on a connection that is already blocked is declined with PublishingPausedError rather than written, so this is the difference between an operator seeing "the broker is out of disk" and seeing a service whose publishes are being declined.

Free: a flag the broker pushed, read out of memory. Unlike #health it costs no round trip, so it is fine to ask per request.

It is not a reason to fail a readiness probe, and Health deliberately does not: see there for why restarting a producer into a broker that is already under pressure helps nobody.

Returns:

  • (String, nil) —

    the broker's own words, "low on disk space" or "low on memory", or nil when it is not blocked and nil when the transport cannot say



494
 
# File 'lib/acemq/amqp/connection.rb', line 494

def blocked_reason = Health.blocked_reason(@transport)

#close(timeout: DRAIN_TIMEOUT) ⇒ nil

Stops every consumer and closes the connection, within one deadline.

Consumers are stopped first and their handlers allowed to finish, so a message being worked on when this is called is acknowledged rather than returned to the queue for somebody else to redo.

The deadline is for the whole drain, not for each consumer. That distinction is the entire point of the keyword: a per-consumer wait is not a bound on anything a process can be held to, because eight consumers each given thirty seconds is four minutes, and four minutes into a shutdown Kubernetes has long since sent SIGKILL — killing every handler mid-flight and leaving everything it was holding unsettled, which is the exact outcome draining exists to avoid. Every consumer is stopped at once, and then all of their handlers are waited for against the one deadline.

When the deadline expires with handlers still running, this closes their channels anyway and raises DrainTimeout once the socket is shut. Nothing is lost — an unacknowledged delivery goes back to the broker and is redelivered — but the drain did not do what it was asked, and saying so is the honest answer. Reporting success would mean an operator whose grace period is too short never finds out. The handlers themselves run on to the end and their outcome is discarded; AceMQ::AMQP::Consumer#cancel says why.

Parameters:

  • timeout (Numeric) (defaults to: DRAIN_TIMEOUT) —

    seconds for the whole drain; see DRAIN_TIMEOUT for why twenty. Zero stops every consumer without waiting for any handler

Returns:

  • (nil)

Raises:

  • (DrainTimeout) —

    when the deadline expired with handlers running



568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
 
# File 'lib/acemq/amqp/connection.rb', line 568

def close(timeout: DRAIN_TIMEOUT)
  consumers = @lock.synchronize do
    taken = @consumers
    @consumers = []
    taken
  end

  failure, stranded = Drain.of(consumers, timeout)
  @transport.close
  # The transport is closed before either of these is raised, and that
  # order is deliberate: a shutdown that reports a problem having left the
  # socket open is a process that will not exit.
  raise failure if failure
  raise DrainTimeout.new(stranded, timeout) unless stranded.empty?

  nil
end

#consume(queue, codec: nil, retry_policy: nil, prefetch: nil, concurrency: 1, tag: nil, arguments: {}, retry_threshold: nil, dead_letters: true, &handler) ⇒ Consumer

Reads messages from a queue until the returned consumer is cancelled.

consumer = mq.consume("orders.new") do |message|
place(message.payload)
Ack.accept
end

The handler is called on the transport's threads, not this one, and must return an Ack. A handler that raises is treated as Ack.retry, except for FatalError, which is treated as Ack.reject — raising that is how a handler says "stop now" without having to know how many attempts remain.

Parameters:

  • queue (String)
  • concurrency (Integer) (defaults to: 1) —

    messages worked on at once. One by default, which keeps a queue's messages in the order the broker offers them; raising it trades that order for throughput.

  • retry_threshold (Numeric, nil) (defaults to: nil) —

    seconds, or nil for the connection's. A retry delayed this long or longer waits in a rung queue instead of in this process.

  • dead_letters (Boolean) (defaults to: true) —

    false declares no {queue}.dlq and {queue}.parked. Only for a handler that accepts everything, as the request-reply reply consumer does: a message given up on then has nowhere to land and is dropped. @api private

Returns:



440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
 
# File 'lib/acemq/amqp/connection.rb', line 440

def consume(queue, codec: nil, retry_policy: nil, prefetch: nil, concurrency: 1,
            tag: nil, arguments: {}, retry_threshold: nil, dead_letters: true,
            &handler)
  unless handler
    raise ArgumentError,
          "consume(#{queue.inspect}) needs a block to handle messages"
  end

  consumer = Consumer.new(
    transport: @transport, queue: queue, handler: handler,
    codec: codec.nil? ? @codec : Codec.check!(codec),
    retry_policy: retry_policy || @retry_policy,
    retry_threshold: retry_threshold || @retry_threshold,
    interceptors: @interceptors, telemetry: @telemetry
  )
  consumer.start(prefetch: prefetch || @prefetch, concurrency: concurrency, tag: tag,
                 arguments: arguments, dead_letters: dead_letters)
  @lock.synchronize { @consumers << consumer }
  consumer
end

#consumers ⇒ Object

The consumers started on this connection, as they stand.

A copy, taken under the lock. Handing back the list itself would let a caller iterate it while a consumer is being added on another thread, which is the sort of bug that only appears under load.



504
 
# File 'lib/acemq/amqp/connection.rb', line 504

def consumers = @lock.synchronize { @consumers.dup }

#declare_exchange(name, **options) ⇒ Object 



506
 
# File 'lib/acemq/amqp/connection.rb', line 506

def declare_exchange(name, **options) = @transport.declare_exchange(name, **options)

#declare_queue(name, queue_type: nil, durable: true, auto_delete: false, exclusive: false, arguments: {}) ⇒ Object

Declares a durable quorum queue, which is what a queue is here unless something says otherwise.

The same default as Java's declareQueue, and the reason is interop rather than taste: x-queue-type is part of a queue's identity to the broker, so two services sharing orders have to declare the same kind or the second one is refused with PRECONDITION_FAILED and cannot consume at all. QueueType has the rest of the rule — a queue that is exclusive, auto-deleting or transient stays classic because RabbitMQ allows nothing else, and a x-queue-type already in arguments is honoured, which is how a stream is declared.

Parameters:

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

    :classic, :quorum or :stream

Raises:



522
523
524
525
526
527
528
529
530
 
# File 'lib/acemq/amqp/connection.rb', line 522

def declare_queue(name, queue_type: nil, durable: true, auto_delete: false,
                  exclusive: false, arguments: {})
  type, arguments = QueueType.resolve(name: name, requested: queue_type,
                                      durable: durable, exclusive: exclusive,
                                      auto_delete: auto_delete, arguments: arguments)
  @transport.declare_queue(name, queue_type: type, durable: durable,
                                 auto_delete: auto_delete, exclusive: exclusive,
                                 arguments: arguments)
end

#delete_queue(name) ⇒ Object 



536
 
# File 'lib/acemq/amqp/connection.rb', line 536

def delete_queue(name) = @transport.delete_queue(name)

#health ⇒ Health::Report

Whether this process can still do what it is running to do.

Costs a round trip to the broker, so it belongs on a readiness probe rather than in a request. See Health for what the statuses mean and why a stopped consumer is degraded rather than down.

Returns:



474
 
# File 'lib/acemq/amqp/connection.rb', line 474

def health = Health.of(self)

#intercept_consume(interceptor = nil, order: nil, &block) ⇒ Connection

Adds an interceptor to every consumer on this connection.

mq.intercept_consume(Tracing.new)

Parameters:

  • interceptor (#before_handle, nil) (defaults to: nil)
  • order (Integer, nil) (defaults to: nil) —

    lower runs first on the way in, last on the way out

Returns:



264
265
266
267
 
# File 'lib/acemq/amqp/connection.rb', line 264

def intercept_consume(interceptor = nil, order: nil, &block)
  @interceptors.add_consume(interceptor, order: order, &block)
  self
end

#intercept_publish(interceptor = nil, order: nil, &block) ⇒ Connection

Adds an interceptor to every publish on this connection.

mq.intercept_publish { |context| context.set_header("tenant", Current.tenant) }

See Interceptors for the hooks an object may answer, what raising from each one means, and how order decides which runs first.

Registration is expected at start-up. A publisher reads the list at the moment it publishes rather than copying it, so an interceptor added later does apply to publishers that already exist — but a message already on its way will not see it.

Parameters:

  • interceptor (#before_publish, nil) (defaults to: nil)
  • order (Integer, nil) (defaults to: nil) —

    lower runs first

Returns:



251
252
253
254
 
# File 'lib/acemq/amqp/connection.rb', line 251

def intercept_publish(interceptor = nil, order: nil, &block)
  @interceptors.add_publish(interceptor, order: order, &block)
  self
end

#message_count(queue) ⇒ Object 



534
 
# File 'lib/acemq/amqp/connection.rb', line 534

def message_count(queue) = @transport.message_count(queue)

#publish(payload, to:, exchange: "", envelope: nil, codec: nil, persistent: true, reply_to: nil, mandatory: false, **fields) ⇒ Envelope

Publishes one message.

The envelope is built here unless one is passed: an identifier, a type falling back to the routing key, a correlation falling back to the identifier, this connection's origin, and the current time. Envelope fields given as keywords override those.

mq.publish(order, to: "orders.new", type: "order.placed.v2",
         correlation_id: request_id)

Parameters:

  • payload (Object) —

    anything the codec will encode

  • to (String) —

    the routing key, or the queue name when publishing through the default exchange

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

    empty for the default exchange, which routes to the queue whose name matches the routing key

  • envelope (Envelope, nil) (defaults to: nil) —

    one built elsewhere, for when a message's metadata derives from another message's

  • reply_to (String, nil) (defaults to: nil) —

    AMQP's own reply-to property, for a message that is asking a question. Not an envelope field, because it is not a header: it is the property Java and .NET responders read, and Patterns::Requester writes it alongside the acemq-reply-to header so that a responder in any of the five libraries can answer.

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

    whether a message that reaches no queue at all is an error rather than a silence. False by default, which is what the broker does on its own and what Go and Python default to: a publish to an exchange with no matching binding is confirmed and dropped, and nothing anywhere says so. Pass true and the broker hands the message back instead, which raises a PublishError whose PublishError#unroutable? is true, counts acemq.publish.total{outcome="unroutable"}, and marks the publish span unroutable. It costs a round trip only when a message really did go nowhere.

  • fields (Hash) —

    envelope fields, when no envelope is given

Returns:

  • (Envelope) —

    what was actually put on the wire, which is what the interceptors left rather than what was handed in

Raises:

  • (PublishingPausedError) —

    when the broker has blocked the connection: nothing was sent, and sending again once it clears is safe

  • (PublishError) —

    when the broker would not take the message, or when mandatory and it reached no queue



318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
 
# File 'lib/acemq/amqp/connection.rb', line 318

def publish(payload, to:, exchange: "", envelope: nil, codec: nil, persistent: true,
            reply_to: nil, mandatory: false, **fields)
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  if envelope && !fields.empty?
    raise ArgumentError,
          "publish was given both an envelope and the fields to build one " \
          "(#{fields.keys.join(", ")}); pass one or the other"
  end

  codec = codec.nil? ? @codec : Codec.check!(codec)
  envelope ||= Envelope.new(origin: @origin, **fields)
  # Built whether or not anything is registered. It is one small object
  # against a round trip to a broker, and the alternative is two code
  # paths through the one method every message in the process goes down.
  context = PublishContext.new(exchange: exchange, routing_key: to,
                               envelope: envelope, payload: payload,
                               reply_to: reply_to, mandatory: mandatory)
  send_intercepted(context, codec, persistent, started)
rescue StandardError => e
  # Counted before the interceptors are told, so a publish refused by an
  # interceptor is counted too. It did not reach the broker, which is the
  # thing this metric is about.
  #
  # +unroutable+ rather than +failed+ for a message the broker took and
  # could not route, because those are two different problems: a failed
  # publish is a broker or a network, an unroutable one is a binding
  # nobody made, and an operator told +failed+ goes looking in the wrong
  # place. The span reads the same word off the same failure.
  record_publish(context, Telemetry::Outcome.of_publish_failure(e), started,
                 exchange: exchange, to: to)
  @interceptors.on_publish_error(context, e) if context
  raise
end

#publish_all(payloads, to:, exchange: "", envelopes: nil, codec: nil, persistent: true, reply_to: nil, mandatory: false, **fields) ⇒ Array<Envelope>

Publishes a batch, and waits for every confirm together.

What bulk publishing actually wants: the throughput of pipelining with the safety of having waited. Every message is handed to the broker before any confirm is asked for, and then all of them are waited on at once. A loop around #publish is the other thing — a round trip to the broker and back per message — and it is the thing a caller can already write, which is why this method exists at all.

mq.publish_all(orders, to: "order.placed", exchange: "orders-events")

It is not atomic. AMQP has no such thing: there is no way to publish a hundred messages such that all or none of them arrive, and a library offering one would be lying. A batch that half arrived is the ordinary outcome of a broker problem partway through, so the failure says how many did arrive rather than only that something went wrong — a caller told nothing but "it failed" republishes messages that are already on a queue.

Every message goes through the publish interceptors exactly as a single publish does, once each, and mandatory: means for each message what it means for one: a message the broker had nowhere to route is that message's failure, carrying PublishError#unroutable?, and the batch counts it among the ones that did not arrive.

Nothing here bounds how many messages may be unconfirmed at once — the batch is as big as the array handed in. A batch of a hundred thousand is a hundred thousand messages held in memory by this process and by the broker, so split a very large one yourself.

Parameters:

  • payloads (Array<Object>) —

    anything the codec will encode, in the order they should be sent

  • to (String) —

    the routing key every message is sent with

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

    empty for the default exchange

  • envelopes (Array<Envelope>, nil) (defaults to: nil) —

    one envelope per payload, for messages whose metadata derives from other messages. One built here per payload when it is not given, which is what #publish does for a single message

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

    whether a message that reaches no queue at all is an error rather than a silence, per message

  • fields (Hash) —

    envelope fields, when no envelopes are given

Returns:

  • (Array<Envelope>) —

    what actually went on the wire, in the order the payloads were given — whatever order the broker confirmed them in

Raises:

  • (PublishError) —

    when any message was not confirmed. The message says how many failed and how many did not, and quotes the first failure in payload order, which is not necessarily the first one the broker answered. A PublishingPausedError when every failure was a message declined unsent because the broker had blocked the connection



400
401
402
403
404
405
406
407
408
409
410
411
412
413
 
# File 'lib/acemq/amqp/connection.rb', line 400

def publish_all(payloads, to:, exchange: "", envelopes: nil, codec: nil, persistent: true,
                reply_to: nil, mandatory: false, **fields)
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  payloads = check_batch!(payloads, envelopes, fields)
  return [] if payloads.empty?

  codec = codec.nil? ? @codec : Codec.check!(codec)
  contexts = payloads.each_with_index.map do |payload, index|
    envelope = envelopes ? envelopes[index] : Envelope.new(origin: @origin, **fields)
    PublishContext.new(exchange: exchange, routing_key: to, envelope: envelope,
                       payload: payload, reply_to: reply_to, mandatory: mandatory)
  end
  send_all_intercepted(contexts, codec, persistent, started)
end

#pull(queue) ⇒ Object 



533
 
# File 'lib/acemq/amqp/connection.rb', line 533

def pull(queue) = @transport.pull(queue)

#queue_exists?(name) ⇒ Boolean

Returns:

  • (Boolean) 


535
 
# File 'lib/acemq/amqp/connection.rb', line 535

def queue_exists?(name) = @transport.queue_exists?(name)