Class: AceMQ::AMQP::Interceptors

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

Overview

The interceptors registered on a connection, and the rules for running them.

The extension point for the things every message in an organisation needs and no library can guess: a tenant identifier, a trace context, a log scope, a size limit, a metric. Without a seam they end up copied into every call site, where one of them is always missing.

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

mq.intercept_consume do |context|
raise AceMQ::AMQP::FatalError, "not our tenant" unless ours?(context)
end

An interceptor is anything answering the hooks it cares about, so the common case is a block and the full case is an object:

class Timing
def order = -100
def before_handle(context) = @started[context.envelope.id] = now
def after_handle(context, ack) = record(context, ack, now - @started.delete(...))
end

mq.intercept_consume(Timing.new)

Nothing here needs anything private. An interceptor is handed a context whose every field is public and is registered through a public method, so everything in this library that could have been an interceptor could have been written by somebody outside it — which is the same rule the patterns follow, and the only way to know a seam is wide enough.

What a failure means

Raising from before_publish stops the publish, and the caller sees the exception. That is the point of intercepting rather than observing: a message that must not go out is stopped once, here, rather than in every publisher.

Raising from before_handle means the handler never runs and the delivery is treated exactly as a failed handler would be — retried, then dead-lettered. An interceptor that refuses a message has to be willing for that message to end up in the dead-letter queue, which is the honest outcome; the alternative is acknowledging something nothing processed. FatalError still means what it means, so an interceptor that refuses for a reason retrying cannot fix can say so.

Raising from after_confirm, after_handle or on_error is reported on stderr and otherwise ignored. By then the message has been sent or the delivery settled, and an exception cannot un-send or un-settle it: letting it out would report a successful publish as a failed one, or skip the remaining teardowns.

Order

Lower order runs first, and interceptors with the same order run in the order they were registered. On the way out of a handler the order is reversed, so a pair that opens something on the way in and closes it on the way out nests properly: the first to open is the last to close.

An interceptor is called from whatever thread is publishing or handling, so one that keeps state has to be safe to call from several at once.

Defined Under Namespace

Classes: Registered

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Interceptors

Returns a new instance of Interceptors.



260
261
262
263
264
265
 
# File 'lib/acemq/amqp/interceptors.rb', line 260

def initialize
  @lock = Mutex.new
  @publishing = [].freeze
  @consuming = [].freeze
  @registered = 0
end

Instance Method Details

#add_consume(interceptor = nil, order: nil) {|context| ... } ⇒ Interceptors

Adds an interceptor to every consumer on this connection.

Parameters:

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

    or nil when a block is given

  • order (Integer, nil) (defaults to: nil) —

    lower runs first; nil asks the interceptor

Yield Parameters:

Returns:



284
285
286
287
 
# File 'lib/acemq/amqp/interceptors.rb', line 284

def add_consume(interceptor = nil, order: nil, &block)
  add(:@consuming, interceptor, block, order,
      before: :before_handle, after: :after_handle)
end

#add_publish(interceptor = nil, order: nil) {|context| ... } ⇒ Interceptors

Adds an interceptor to every publish on this connection.

Parameters:

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

    or nil when a block is given

  • order (Integer, nil) (defaults to: nil) —

    lower runs first; nil asks the interceptor

Yield Parameters:

Returns:



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

def add_publish(interceptor = nil, order: nil, &block)
  add(:@publishing, interceptor, block, order,
      before: :before_publish, after: :after_confirm)
end

#after_confirm(context) ⇒ Object

Runs every after_confirm, once the broker has the message.



310
311
312
313
314
315
 
# File 'lib/acemq/amqp/interceptors.rb', line 310

def after_confirm(context)
  @publishing.each do |entry|
    safely(entry.after, "after a publish") { entry.after.call(context) }
  end
  context
end

#after_handle(context, ack) ⇒ Object

Runs every after_handle, in reverse order, after the handler has returned or raised and before the delivery is settled.

The context carries the Settlement by then, so an interceptor is told what is about to happen rather than only what was asked for.



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

def after_handle(context, ack)
  @consuming.reverse_each do |entry|
    safely(entry.after, "after a handler") { entry.after.call(context, ack) }
  end
  context
end

#before_handle(context) ⇒ Object

Runs every before_handle, in order. An exception is let out and the consumer treats it as a handler failure, which is the honest outcome for a refused message.



328
329
330
331
 
# File 'lib/acemq/amqp/interceptors.rb', line 328

def before_handle(context)
  @consuming.each { |entry| entry.before&.call(context) }
  context
end

#before_publish(context) ⇒ PublishContext

Runs every before_publish, in order. An exception is let straight out: refusing a publish is the whole reason a policy interceptor exists, and swallowing it here would turn "this message is not allowed" into "this message was sent".

Parameters:

Returns:



304
305
306
307
 
# File 'lib/acemq/amqp/interceptors.rb', line 304

def before_publish(context)
  @publishing.each { |entry| entry.before&.call(context) }
  context
end

#consuming ⇒ Object 



295
 
# File 'lib/acemq/amqp/interceptors.rb', line 295

def consuming = @consuming.size

#empty? ⇒ Boolean

Whether anything at all is registered, which is what a caller checks before doing work only an interceptor would want done.

Returns:

  • (Boolean) 


291
 
# File 'lib/acemq/amqp/interceptors.rb', line 291

def empty? = @publishing.empty? && @consuming.empty?

#on_consume_error(context, failure) ⇒ Object

Runs every on_error after a handler raised.



346
347
348
 
# File 'lib/acemq/amqp/interceptors.rb', line 346

def on_consume_error(context, failure)
  report(@consuming, context, failure, "a failed handler")
end

#on_publish_error(context, failure) ⇒ Object

Runs every on_error after a publish failed, including when an interceptor refused it. Never allowed to replace the original failure: the caller needs to see what went wrong, not what an observer did about it.



321
322
323
 
# File 'lib/acemq/amqp/interceptors.rb', line 321

def on_publish_error(context, failure)
  report(@publishing, context, failure, "a failed publish")
end

#publishing ⇒ Object

How many are registered on each side.



294
 
# File 'lib/acemq/amqp/interceptors.rb', line 294

def publishing = @publishing.size