class

LavinMQ::Shovel::Runner

Inherits LavinMQ::Shovel::OutcomeListener < LavinMQ::SortableJSON < Reference < Object

Constants

ABORT_THRESHOLD = 10

MAX_DELAY = 300

MAX_DELIVERY_DELAY = 30

ORIGIN = RoughTime.instant

Monotonic clock as an Int64 so the deadline can live in an Atomic. The coarse clock is plenty for backoff windows of 0.5s and up, and spares a clock read per delivered message.

RETRY_THRESHOLD = 10

Constructors

new(source : Source, destination : Destination, name : String, vhost : VHost, reconnect_delay : Time::Span = DEFAULT_RECONNECT_DELAY)

Source

Class methods

delivery_backoff(failures : Int32) : Time::Span

Backoff before the next delivery attempt after failures consecutive transient failures: 0.5s, 1s, 2s, 4s, … doubling, capped at MAX_DELIVERY_DELAY, then holding there.

Source

Instance methods

aborted?

delete

delete_paused_file

details_tuple

exponential_reconnect_delay

name

pause

paused?

pending_backoff

Time left before the next delivery may be attempted; zero when the destination is healthy.

Source

report(delivery_tag : UInt64, outcome : Outcome)

The single place a delivery Outcome becomes a source action: a confirmed delivery acks; a rejection (broker nack or HTTP error) requeues so queue-based retry/DLX policies still apply. Called by the destination (synchronously for HTTP, from the confirm fiber for AMQP on-confirm).

Source

resume

Start a new run for a paused or aborted shovel.

Source

run

Source

running?

Source

should_stop_loop?(run_generation)

Source

state

Source