Class: AceMQ::AMQP::Transport

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

Overview

The broker, over AMQP 0-9-1.

Everything above this class is about envelopes, codecs and retries and knows nothing about channels; everything below it is bunny. The seam is here so the consumer's retry arithmetic can be tested without a broker, by handing Connection something else that answers these methods.

Defined Under Namespace

Classes: Subscription

Constant Summary collapse

UNEXPLAINED_BLOCK =

What #blocked_reason says when the broker has blocked the connection and this process cannot say why.

Two ways that happens: a session started elsewhere and handed to #initialize after RabbitMQ had already blocked it, so the frame carrying the reason came and went before anything here was listening; or an application that registered its own on_blocked on the session afterwards, which replaces the callback below — bunny keeps one. Neither is worth a silence: that the connection is blocked at all is the half an operator acts on.

"the broker did not say why"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(session, max_outstanding_publishes: PublishPermits::DEFAULT, single_host: false) ⇒ Transport

Returns a new instance of Transport.

Parameters:

  • session (Bunny::Session) —

    an already started connection

  • max_outstanding_publishes (Integer) (defaults to: PublishPermits::DEFAULT) —

    see PublishPermits

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

    whether the session has one broker address, so a recovery can go straight back to it; see #watch_recovery. open works it out; a caller handing in a session of its own says so



1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
 
# File 'lib/acemq/amqp/transport.rb', line 1017

def initialize(session, max_outstanding_publishes: PublishPermits::DEFAULT,
               single_host: false)
  @session = session
  @single_host = single_host
  @permits = PublishPermits.new(max_outstanding_publishes)
  @lock = Mutex.new
  # A lock of its own, because a pull holds messages unacknowledged across
  # a whole pass and settling one has to reach the channel it came down.
  # Sharing the publish lock would mean a replay's own republish waiting
  # on the channel it is about to acknowledge from.
  @pull_lock = Mutex.new
  @subscriptions = []
  @returns = ReturnedMessages.new
  @confirms = ConfirmWait.new
  # A lock of its own again, and the smallest one in the class: the
  # broker's blocked callback runs on bunny's reader thread, and a health
  # probe reads it from a web server's. Sharing the publish lock would
  # mean a readiness probe queuing behind a publish that is itself waiting
  # for the broker that has just blocked it.
  @blocked_lock = Mutex.new
  @blocked_reason = nil
  # Its own lock again, and for the same reason: this one is written from
  # bunny's recovery thread and read by every thread that publishes.
  @recovery_lock = Mutex.new
  @recovering = false
  watch_blocked
  watch_recovery
end

Instance Attribute Details

#session ⇒ Object (readonly)

The bunny session, for the things this class deliberately does not wrap.



1047
1048
1049
 
# File 'lib/acemq/amqp/transport.rb', line 1047

def session
  @session
end

Class Method Details

.default_vhost(url) ⇒ 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.

A URL whose path is a bare slash, with the slash taken off.

amqp://host:5672/ is the URL the Go and Python libraries print in their documentation and connect to the default vhost with, and the one the TCK hands every driver. Bunny reads the empty path after that slash as a vhost named "", which the broker refuses with NOT_ALLOWED, so the same URL worked there and not here. Without the slash bunny uses "/", as they do; a vhost named in the path, %2F included, is left alone.



988
989
990
991
992
 
# File 'lib/acemq/amqp/transport.rb', line 988

def self.default_vhost(url)
  return url unless url.is_a?(String)

  url.sub(%r{\A(amqps?://[^/?#]+)/(?=[?#]|\z)}i, '\1')
end

.load_driver! ⇒ Object

How this transport reaches RabbitMQ.

Required lazily, and only here. The gem declares no runtime dependencies because reading an AceMQ envelope should not oblige a process to install a broker client it will never open a socket with — an audit tool that parses headers out of a log has no business compiling in an AMQP stack. The cost of that choice is that the failure moves from install time to the first connection, so it is caught here and re-raised saying what to install.



913
914
915
916
917
918
919
920
 
# File 'lib/acemq/amqp/transport.rb', line 913

def self.load_driver!
  require "bunny"
rescue LoadError => e
  raise DependencyMissing,
        "the AceMQ transport needs the bunny gem, which is not installed. " \
        "Add `gem \"bunny\", \"~> 2.23\"` to your Gemfile, or run " \
        "`gem install bunny`. (#{e.message})"
end

.open(url, heartbeat: :server, connection_timeout: 10, security: nil, credentials: nil, max_outstanding_publishes: PublishPermits::DEFAULT, **options) ⇒ Transport

Opens a connection.

The URL decides how the connection is protected unless a Security says otherwise: amqps:// is encrypted and the broker is verified against the machine's trust store, amqp:// is plaintext. Pass security: for a private certificate authority, a client certificate, or the deliberate absence of verification; pass credentials: to keep the password out of the URL, and so out of the error message two lines below this one.

The security options are merged over anything in options rather than under it. A caller who reaches past Security to bunny's own TLS keys is describing the same thing twice, and of the two answers the one that went through the checks in this library is the one to honour.

Parameters:

  • url (String) —

    amqp:// or amqps://

  • heartbeat (Integer, Symbol) (defaults to: :server) —

    seconds, or :server to take the broker's suggestion

  • connection_timeout (Numeric) (defaults to: 10) —

    seconds to wait for the handshake

  • security (Security, nil) (defaults to: nil) —

    how to protect the connection

  • credentials (Credentials, #call, nil) (defaults to: nil) —

    the broker login

  • max_outstanding_publishes (Integer) (defaults to: PublishPermits::DEFAULT) —

    how many messages may be waiting for a confirm on this connection at once; see PublishPermits. The same ceiling as Java's maxOutstandingPublishes and .NET's MaxOutstandingPublishes, and the same default

  • options (Hash) —

    anything else bunny understands

Returns:

Raises:



952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
 
# File 'lib/acemq/amqp/transport.rb', line 952

def self.open(url, heartbeat: :server, connection_timeout: 10, security: nil,
              credentials: nil, max_outstanding_publishes: PublishPermits::DEFAULT,
              **options)
  load_driver!
  url = default_vhost(url)
  security = Security.for_connection(url, security: security, credentials: credentials)
  # Built before the socket is opened: a ceiling of nought is a
  # configuration mistake, and finding it out after a connection has been
  # established means an error that mentions the broker for a reason that
  # has nothing to do with it.
  permits = PublishPermits.new(max_outstanding_publishes)
  # Ours goes first so a caller's own `topology_recovery_filter` in +options+
  # replaces it rather than being silently overridden.
  session = Bunny.new(url, topology_recovery_filter: RecoverLiveChannelsOnly.new,
                           heartbeat: heartbeat, connection_timeout: connection_timeout,
                           **options, **security.to_transport_options)
  security.configure(session)
  session.start
  new(session, max_outstanding_publishes: permits.limit,
               single_host: single_host?(options))
rescue DependencyMissing, ConfigurationError
  raise
rescue StandardError => e
  raise TransportError, "cannot reach the broker at #{redact(url)}: #{e.message}"
end

.redact(url) ⇒ 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.

A URL with its password taken out, for an error message that will be logged. A credential that reaches a log is a credential that has to be rotated.



1008
1009
1010
 
# File 'lib/acemq/amqp/transport.rb', line 1008

def self.redact(url)
  url.to_s.sub(%r{(://[^:/@]+):[^@]*@}, '\1:***@')
end

.single_host?(options) ⇒ Boolean

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.

Whether bunny will have one address to connect to, read the way bunny's own addresses_from reads it: the first of these it finds, or the URL.

Returns:

  • (Boolean) 


998
999
1000
1001
 
# File 'lib/acemq/amqp/transport.rb', line 998

def self.single_host?(options)
  hosts = options[:host] || options[:hostname] || options[:addresses] || options[:hosts]
  Array(hosts).size <= 1
end

Instance Method Details

#bind(queue:, exchange:, routing_key: "") ⇒ Object

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



1084
1085
1086
1087
1088
 
# File 'lib/acemq/amqp/transport.rb', line 1084

def bind(queue:, exchange:, routing_key: "")
  with_admin_channel do |channel|
    channel.queue_bind(queue, exchange, routing_key: routing_key)
  end
end

#blocked? ⇒ Boolean

Whether the broker has asked this connection to stop publishing.

Returns:

  • (Boolean) 


1333
 
# File 'lib/acemq/amqp/transport.rb', line 1333

def blocked? = !blocked_reason.nil?

#blocked_reason ⇒ String?

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

RabbitMQ sends connection.blocked when it is low on memory or disk and connection.unblocked when the alarm clears. Between the two the broker stops reading, so #publish declines with PublishingPausedError rather than writing into a socket nobody reads — and this is the difference between an operator seeing "the broker is out of disk" and seeing a service whose publishes are being declined.

The reason is the broker's own words, "low on disk space" or "low on memory", recorded from the frame that carried it. Bunny keeps the flag but not the reason, so both are asked for: bunny's blocked? decides whether, which means a reason recorded before an automatic recovery cannot outlive the connection it belonged to, and the callback below supplies why.

Returns:

  • (String, nil) 


1326
1327
1328
1329
1330
 
# File 'lib/acemq/amqp/transport.rb', line 1326

def blocked_reason
  return nil unless @session.respond_to?(:blocked?) && @session.blocked?

  @blocked_lock.synchronize { @blocked_reason } || UNEXPLAINED_BLOCK
end

#close ⇒ Object

Cancels every subscription and closes the connection.



1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
 
# File 'lib/acemq/amqp/transport.rb', line 1336

def close
  # Taken out under the lock and cancelled outside it. A handler still
  # running is about to want this same lock to publish a dead letter, and
  # holding it across a cancel that waits on the broker is how a shutdown
  # becomes a deadlock nobody can reproduce.
  subscriptions = @lock.synchronize do
    taken = @subscriptions
    @subscriptions = []
    taken
  end
  subscriptions.each(&:cancel)

  @lock.synchronize do
    @publish_channel.close if @publish_channel&.open?
    @publish_channel = nil
  end
  # Anything a pass was still holding goes back to its queue when this
  # closes, which is the whole reason a declined message is held rather
  # than returned one at a time.
  @pull_lock.synchronize do
    @pull_channel.close if @pull_channel&.open?
    @pull_channel = nil
  end
  @session.close
  nil
rescue StandardError => e
  raise TransportError, "cannot close the connection cleanly: #{e.message}"
end

#declare_exchange(name, kind:, durable: true, auto_delete: false, arguments: {}) ⇒ Object

Creates an exchange unless it is already there.

Parameters:

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

    direct, topic, fanout or headers



1057
1058
1059
1060
1061
1062
 
# File 'lib/acemq/amqp/transport.rb', line 1057

def declare_exchange(name, kind:, durable: true, auto_delete: false, arguments: {})
  with_admin_channel do |channel|
    channel.exchange_declare(name, kind.to_s, durable: durable, auto_delete: auto_delete,
                                              arguments: stringify(arguments))
  end
end

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

Creates a queue unless it is already there.

Deliberately without the quorum default Connection#declare_queue has: this is the layer that does what it is told, and a caller who reached past the connection to it is declaring exactly what they wrote down. queue_type is honoured when it is given, because AceMQ::AMQP::Topology#apply can be handed a transport and its plan says which kind each queue is.

Parameters:

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

    :classic, :quorum or :stream; nothing is added to the arguments when it is not given



1074
1075
1076
1077
1078
1079
1080
1081
 
# File 'lib/acemq/amqp/transport.rb', line 1074

def declare_queue(name, queue_type: nil, durable: true, auto_delete: false,
                  exclusive: false, arguments: {})
  arguments = QueueType.table(queue_type, arguments) if queue_type
  with_admin_channel do |channel|
    channel.queue_declare(name, durable: durable, auto_delete: auto_delete,
                                exclusive: exclusive, arguments: stringify(arguments))
  end
end

#delete_exchange(name) ⇒ Object

Removes an exchange.



1302
1303
1304
 
# File 'lib/acemq/amqp/transport.rb', line 1302

def delete_exchange(name)
  with_admin_channel { |channel| channel.exchange_delete(name) }
end

#delete_queue(name) ⇒ Object

Removes a queue and every message still on it.

For tests and for tools. A service that deletes queues has usually confused a queue with a session, and the messages go with it — including the ones somebody was about to be paid for.



1297
1298
1299
 
# File 'lib/acemq/amqp/transport.rb', line 1297

def delete_queue(name)
  with_admin_channel { |channel| channel.queue_delete(name) }
end

#max_outstanding_publishes ⇒ Object

How many publishes may be waiting for a confirm on this connection at once. See PublishPermits.



1051
 
# File 'lib/acemq/amqp/transport.rb', line 1051

def max_outstanding_publishes = @permits.limit

#message_count(queue) ⇒ Object

How many messages are waiting on a queue.

A number for a dashboard or a test, not a decision to make in a handler: it is a snapshot of a queue that is still moving, and it is already wrong by the time it is read anywhere else.



1277
1278
1279
1280
1281
 
# File 'lib/acemq/amqp/transport.rb', line 1277

def message_count(queue)
  with_admin_channel do |channel|
    channel.queue_declare(queue, passive: true).message_count
  end
end

#open? ⇒ Boolean

Whether the connection is up.

Returns:

  • (Boolean) 


1307
 
# File 'lib/acemq/amqp/transport.rb', line 1307

def open? = @session.open?

#publish(exchange:, routing_key:, body:, content_type: nil, message_id: nil, headers: {}, persistent: true, reply_to: nil, mandatory: false) ⇒ String

Sends one message and waits for the broker to take responsibility.

The wait is the point. Without publisher confirms a successful publish means the bytes reached a socket, which is not the same as the broker having them, and the difference only ever shows up as messages that were never anywhere.

mandatory buys the other half of that. A confirm says the broker has the message; it does not say the message reached a queue, and a publish to an exchange with no matching binding is confirmed and dropped in the same breath. Asking for mandatory makes the broker hand such a message back as basic.return before it confirms it, which is the only way AMQP will tell you. It costs a round trip only when the message really is unroutable.

Parameters:

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

    AMQP's own reply-to property, left off the message entirely when it is nil

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

    whether reaching no queue is an error rather than a silence

Returns:

  • (String) —

    the message id it went out with

Raises:



1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
 
# File 'lib/acemq/amqp/transport.rb', line 1115

def publish(exchange:, routing_key:, body:, content_type: nil, message_id: nil,
            headers: {}, persistent: true, reply_to: nil, mandatory: false)
  refuse_while_blocked!(message_id, exchange, routing_key)
  returned = nil
  confirmed = publish_channel([message_id, exchange, routing_key]) do |channel|
    @returns.arm(channel, exchange) if mandatory
    ok = write_and_wait(channel, body.to_s, exchange, routing_key,
                        content_type: content_type, message_id: message_id,
                        reply_to: presence(reply_to), mandatory: mandatory,
                        headers: stringify(headers), persistent: persistent)
    # The return, when there is one, arrives ahead of the confirm and is
    # dispatched on the reader thread while this one waits — so by the
    # time the wait is over it is already waiting to be taken.
    returned = @returns.take if mandatory
    ok
  end
  unless confirmed
    raise PublishError, PublishFailure.not_confirmed(message_id, exchange, routing_key)
  end
  return message_id if returned.nil?

  raise PublishError.new(
    PublishFailure.not_routed(message_id, exchange, routing_key, returned),
    unroutable: true
  )
rescue PublishError
  raise
rescue StandardError => e
  raise PublishError,
        PublishFailure.not_published(message_id, exchange, routing_key, e.message)
end

#publish_all(messages) ⇒ Array<String, PublishError>

Sends a batch, and waits for every confirm at the end.

The difference from a loop around #publish is the whole point of the method: there, every message costs a round trip to the broker and back before the next one is written, because every message waits for its own confirm. Here every message is handed to the channel first and the wait happens once, for all of them — which is what publisher confirms were designed for and what makes a thousand-message publish take about as long as one.

It is not atomic, and nothing in AMQP could make it so. A batch that half arrived is the ordinary outcome of a broker problem partway through, so each message gets its own answer rather than the batch getting one between them.

Parameters:

  • messages (Array<Hash>) —

    the keywords #publish takes, one hash per message

Returns:

  • (Array<String, PublishError>) —

    one entry per message, in the order they were given: the id it went out with, or the failure it met



1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
 
# File 'lib/acemq/amqp/transport.rb', line 1166

def publish_all(messages)
  return [] if messages.empty?

  # Answered per message rather than raised, because that is what this
  # method promises for every other failure.
  if (reason = blocked_reason)
    return messages.map do |m|
      PublishingPausedError.new(
        PublishFailure.paused(m[:message_id], m[:exchange], m[:routing_key], reason)
      )
    end
  end

  batch = BatchPublish.new(messages, @returns, @confirms)
  publish_channel { |channel| batch.run(channel, @permits) }
rescue PublishingPausedError => e
  # Refused before anything was written -- a recovery in progress -- so
  # every message gets the same answer, as it would for a blocked broker.
  messages.map do |m|
    PublishingPausedError.new(
      PublishFailure.not_published(m[:message_id], m[:exchange], m[:routing_key],
                                   e.message)
    )
  end
end

#pull(queue) ⇒ Delivery?

Takes one message off a queue, without subscribing to it.

A subscription is a standing arrangement; this is a single read, which is what a pass over a queue with a beginning and an end needs. Draining a dead-letter queue with a consumer means writing the code that decides when to stop, and getting it wrong means a tool that never exits.

The delivery is unacknowledged, and settling it is the caller's job. That is deliberate: a message left unacknowledged is held by the broker rather than lost, so a tool that dies half way through a pass returns everything it was holding.

Parameters:

  • queue (String)

Returns:

  • (Delivery, nil) —

    nil when the queue has nothing waiting



1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
 
# File 'lib/acemq/amqp/transport.rb', line 1260

def pull(queue)
  @pull_lock.synchronize do
    channel = pull_channel
    info, properties, body = channel.basic_get(queue, manual_ack: true)
    next nil if info.nil?

    delivery_from(channel, info, properties, body, lock: @pull_lock)
  end
rescue StandardError => e
  raise TransportError, "cannot read a message from #{queue.inspect}: #{e.message}"
end

#queue_exists?(name) ⇒ Boolean

Whether a queue is on the broker, creating nothing.

Its own question because a declaration cannot answer it: declaring a queue that is missing creates it, so anything built on declarations would create the very queues it was asked only to look for.

Returns:

  • (Boolean) 


1288
1289
1290
 
# File 'lib/acemq/amqp/transport.rb', line 1288

def queue_exists?(name)
  @session.queue_exists?(name)
end

#subscribe(queue, prefetch: 20, concurrency: 1, tag: nil, arguments: {}) {|delivery| ... } ⇒ Subscription

Delivers messages until the returned subscription is cancelled.

The block runs on bunny's consumer pool rather than on the caller's thread, so concurrency is how many messages this subscription works on at once. One by default, which keeps a queue's messages in the order the broker offers them.

Yield Parameters:

Returns:



1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
 
# File 'lib/acemq/amqp/transport.rb', line 1201

def subscribe(queue, prefetch: 20, concurrency: 1, tag: nil, arguments: {}, &handler)
  # No shutdown timeout on the channel's work pool, and that +nil+ is what
  # keeps a drain inside its deadline.
  #
  # Cancelling the last consumer on a bunny channel shuts its work pool down
  # and, given a timeout, waits that long for a busy worker -- sixty seconds
  # by default -- inside +basic_cancel+, before {Connection#close} has looked
  # at its own deadline at all. Measured on 0.7.5: close(timeout: 0.5) on a
  # 1.5s handler took 1.5s and raised nothing; on a 63s handler it took 60.5s
  # and then said "within 0.5s". With +nil+ the pool is told to stop and the
  # cancel returns, so the only wait left is the library's own, against one
  # deadline. The fourth positional argument has meant this since bunny 2.x
  # and still does in 3.4.
  channel = @session.create_channel(nil, concurrency, false, nil)
  channel.prefetch(prefetch) if prefetch.positive?
  # A consumer tag of our own when the caller gives none, rather than leaving it
  # to the broker.
  #
  # bunny 3.x records every consumer for topology recovery and keys the record by
  # tag, and recovery re-subscribes -- which records the consumer again. With a
  # broker-assigned tag the new registration lands under a *new* key, so the
  # registry grows instead of being replaced: measured at 1, 2, 4 recorded
  # consumers over three recoveries. Each recorded consumer then gets its own
  # `maybe_reinitialize_consumer_pool!`, and all but the last of those pools is
  # orphaned with its threads still parked -- which is the thread growth a soak
  # found in the Ruby standing load. A plain bunny consumer does not grow this
  # way, which is what showed the tag was ours to fix.
  #
  # Generated once here, so it survives every recovery of this subscription and
  # the record is replaced rather than added to.
  tag ||= "acemq-#{queue}-#{SecureRandom.hex(8)}"
  arguments = stringify(arguments)
  position = StreamPosition.for(arguments)
  consumer = channel.queue(queue, passive: true).subscribe(
    manual_ack: true, block: false, consumer_tag: tag, arguments: arguments
  ) do |info, properties, body|
    delivery = delivery_from(channel, info, properties, body)
    handler.call(position ? position.track(delivery) : delivery)
  end
  track(Subscription.new(channel, consumer, position))
rescue StandardError => e
  channel.close if channel&.open?
  raise TransportError, "cannot consume from #{queue.inspect}: #{e.message}"
end