Class: AceMQ::AMQP::Consumer

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

Overview

A running subscription. Cancel it to stop.

Where the retry policy is actually enforced: the attempt is advanced here, the decision about where the delay is waited is made here, and so is the decision to give up and dead-letter — that last one because the broker cannot write x-acemq-error onto a message explaining why it gave up.

A short delay is waited here, holding one prefetch slot. A long one is waited by the broker, in a rung queue; see RetryLadder for why the two are not the same choice.

Constant Summary collapse

ALSO_COUNTED =

The counters a delivery raises besides acemq.consume.total, keyed by the word the Settlement carries.

A retry and a message set aside are counted twice on purpose: once in acemq.consume.total tagged with the outcome, which is the series saying exactly one thing happened to this message, and once under a name of their own, because a retry rate and a dead-letter rate are the two most often wanted without a tag filter. All five languages keep the same pair. Nothing else has a second name; the tag is how they are read.

parked shares dead.lettered.total with dead_lettered rather than having a counter of its own, tagged apart by the outcome. That is Java's rule, and its reasoning is worth repeating: both are a message set aside, and an operator asking "how much is this queue giving up on" wants one number that can then be split.

{ Settlement::RETRIED => Telemetry::RETRIED_TOTAL,
Settlement::DEAD_LETTERED => Telemetry::DEAD_LETTERED_TOTAL,
Settlement::PARKED => Telemetry::DEAD_LETTERED_TOTAL }.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(transport:, queue:, handler:, codec:, retry_policy:, retry_threshold: RetryLadder::DEFAULT_THRESHOLD, interceptors: Interceptors.new, telemetry: nil) ⇒ Consumer

Returns a new instance of Consumer.



779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
 
# File 'lib/acemq/amqp/connection.rb', line 779

def initialize(transport:, queue:, handler:, codec:, retry_policy:,
               retry_threshold: RetryLadder::DEFAULT_THRESHOLD,
               interceptors: Interceptors.new, telemetry: nil)
  @transport = transport
  @queue = queue
  @handler = handler
  @codec = codec
  @interceptors = interceptors
  @telemetry = Telemetry::Reporter.for(telemetry)
  @retry_policy = retry_policy
  @ladder = RetryLadder.for(queue, retry_policy, threshold: retry_threshold)
  @dead_letter_queue = Naming.dead_letter_queue(queue)
  @parked_queue = Naming.parked_queue(queue)
  @lock = Mutex.new
  @in_flight = 0
  @rungs_seen = {}
end

Instance Attribute Details

#codec ⇒ Object (readonly)

Returns the value of attribute codec.



777
778
779
 
# File 'lib/acemq/amqp/connection.rb', line 777

def codec
  @codec
end

#ladder ⇒ Object (readonly)

Returns the value of attribute ladder.



777
778
779
 
# File 'lib/acemq/amqp/connection.rb', line 777

def ladder
  @ladder
end

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



777
778
779
 
# File 'lib/acemq/amqp/connection.rb', line 777

def queue
  @queue
end

#retry_policy ⇒ Object (readonly)

Returns the value of attribute retry_policy.



777
778
779
 
# File 'lib/acemq/amqp/connection.rb', line 777

def retry_policy
  @retry_policy
end

Instance Method Details

#cancel(timeout: Connection::DRAIN_TIMEOUT) ⇒ nil

Stops delivery and waits for handlers already running.

The subscription is stopped and closed whether or not the wait ran out, because a consumer that has been cancelled and is still subscribed is the worst of both: it is not draining and it is still being sent work. Whatever a handler was holding when time ran out is unacknowledged, so the broker redelivers it. #in_flight afterwards is how many that was, and it is what AceMQ::AMQP::Connection#close reports in its DrainTimeout.

A handler still running at the deadline is left to finish, and its outcome is thrown away. Ruby cannot stop a thread safely — Thread#kill in the middle of a database transaction is worse than anything it would save — so the thread runs on, as Java's drain leaves it ("the consumer is still stopped, but something is still running"). Whatever it returns is not settled: no ack, no retry, no dead letter. The channel it came down is already closed, so the broker has requeued the message and will redeliver it; settling it as well would be the same message twice.

Deliveries bunny had already received but no handler had started are not started at all once this is called. They go back with the channel.

Parameters:

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

    seconds to wait for in-flight handlers. Zero stops the consumer without waiting at all

Returns:

  • (nil) 


907
908
909
910
 
# File 'lib/acemq/amqp/connection.rb', line 907

def cancel(timeout: Connection::DRAIN_TIMEOUT)
  failure, = Drain.of([self], timeout)
  raise failure if failure
end

#handle(delivery) ⇒ Object

Runs one delivery through the codec, the handler and the retry policy.

Public because it is the whole of this class worth testing, and a test that has to open a broker to reach it is a test nobody runs.



834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
 
# File 'lib/acemq/amqp/connection.rb', line 834

def handle(delivery)
  enter
  envelope = envelope_for(delivery)
  # Which go this delivery is, recorded after the envelope and before the
  # codec: a body that will not decode is still a delivery that arrived,
  # and that ordering is what makes this distribution's sample count
  # trustworthy as "deliveries in".
  @telemetry.observe(Telemetry::CONSUME_ATTEMPTS, envelope.attempt, queue: @queue)
  context = context_for(delivery, envelope, decode(delivery))
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  ack = invoke(context, delivery)
  took = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
  # Decided before the interceptors are told rather than after, because
  # the decision is the thing worth telling them: an ack asking for a
  # retry on the last attempt is a dead letter, and a retry that will
  # really happen has a delay attached that does not exist until the
  # policy has been asked. An interceptor left to infer it from the ack
  # reports the message as retried and the delay as nothing.
  context.settlement = decide(context.envelope, ack)
  observe(context.settlement, took)
  # After the handler and before the delivery is settled, so an
  # interceptor can still see how it went and still undo whatever it set
  # up on the way in. The envelope read back off the context is the one an
  # interceptor may have changed, and it is the one a dead letter is
  # written with — otherwise a header stamped on the way in would be there
  # for the handler and gone from the queue somebody has to look at.
  @interceptors.after_handle(context, ack)
  # Not settled when the drain gave up on this handler: the channel is
  # gone, the broker has the message back, and an ack would fail while a
  # retry or a dead letter would publish a second copy. See {#cancel}.
  settle(delivery, context.envelope, context.settlement) unless @released
rescue DecodeError => e
  # A body that will not decode decodes no better next time, so it is
  # parked rather than retried. Parked and not dead-lettered: a message
  # that failed five times and a message nothing could read are two
  # different problems, and whoever drains the dead letters should not
  # have to sort them by hand.
  #
  # Counted here rather than inside {#park}, because this delivery never
  # reached a handler and so never reached {#observe}: a handler that asks
  # for {Ack.park} is counted there with every other outcome, and counting
  # in both places would count it twice.
  @telemetry.count(Telemetry::CONSUME_TOTAL, 1,
                   queue: @queue, outcome: Telemetry::Outcome::PARKED)
  park(delivery, envelope, "could not be decoded: #{e.message}")
ensure
  leave
end

#in_flight ⇒ Object

How many messages this consumer is working on right now.



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

def in_flight = @lock.synchronize { @in_flight }

#release ⇒ Integer

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.

The last step of #cancel: closes the channel, which hands every unsettled delivery back to the broker, and stops any handler still running from settling.

Returns:

  • (Integer) —

    how many handlers were still running



932
933
934
935
 
# File 'lib/acemq/amqp/connection.rb', line 932

def release
  @released = true
  in_flight.tap { @subscription&.close }
end

#running? ⇒ Boolean

Whether the broker is still sending this consumer messages.

Asked of the subscription rather than remembered here, because the two can disagree: a channel closed by the broker, or taken down by an error on it, stops delivery without anything in this process being told. A flag set in #cancel would say "running" for a consumer that had been deaf for an hour.

Returns:

  • (Boolean) 


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

def running? = !!@subscription&.open?

#start(prefetch:, concurrency:, tag:, arguments:, dead_letters: true) ⇒ 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. 



807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
 
# File 'lib/acemq/amqp/connection.rb', line 807

def start(prefetch:, concurrency:, tag:, arguments:, dead_letters: true)
  # Declared before anything is subscribed, and not on the failure path.
  # A rung, a dead-letter queue or a parking queue that does not exist
  # loses the message rather than reporting anything — the default
  # exchange drops what it cannot route — so the moment to find out is the
  # one where nothing has failed yet.
  #
  # Both halves, and the dead-letter half even for a consumer with no
  # retry policy: this consumer republishes to +{queue}.dlq+ and
  # +{queue}.parked+ by name whatever its policy says, and a topology
  # nobody remembered to apply must not be the difference between a dead
  # letter somebody can read and one that never existed.
  #
  # The one exception is a consumer started with +dead_letters: false+,
  # which is the request-reply reply consumer: it accepts everything, so
  # durable queues hung off an exclusive reply queue would only outlive it.
  @ladder.declare(@transport) if dead_letters
  @subscription = @transport.subscribe(
    @queue, prefetch: prefetch, concurrency: concurrency, tag: tag, arguments: arguments
  ) { |delivery| handle(delivery) unless @stopped }
  self
end

#stop ⇒ 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.

The first step of #cancel: no more deliveries, and none of those bunny has already received is started — they go back with the channel. Returns at once; bunny is not allowed to wait for a busy handler here, see Transport#subscribe.



921
922
923
924
 
# File 'lib/acemq/amqp/connection.rb', line 921

def stop
  @stopped = true
  @subscription&.stop
end