Redis Ack Emulation (kombu.transport.redis.QoS).
Tracks unacknowledged messages in Redis using:
- a Hash (
unacked) mappingdelivery_tag -> JSON([raw_msg, exchange, routing_key]) - a Sorted Set (
unacked_index) mappingdelivery_tag -> unix_timestamp - a Mutex key (
unacked_mutex) for periodic restoration of messages exceedingvisibility_timeout
Summary
Functions
Acknowledges delivery_tag, removing it from Redis unacked / unacked_index and local QoS.
Appends a delivered message to the Redis unacked hash and unacked_index sorted set,
as well as local channel QoS state.
Rejects delivery_tag. If requeue: true, restores the message to the front of its queue (leftmost: true).
Removes delivery_tag from unacked_index and unacked in Redis.
Looks up delivery_tag in unacked, deletes it from unacked and unacked_index,
and invokes restore_fun.(raw_message, exchange, routing_key, leftmost) if found.
Restores all unacknowledged messages tracked by this channel instance.
Restores unacked messages once during channel shutdown if restore_at_shutdown is enabled.
Scans unacked_index for messages older than visibility_timeout and restores them,
holding a distributed Redis mutex (unacked_mutex) during restoration.
Types
@type t() :: %Kombu.Transport.Redis.QoS{ client: Kombu.Transport.Redis.PrefixedClient.t() | nil, last_restored: number() | nil, restore_at_shutdown: boolean(), unacked_index_key: String.t(), unacked_key: String.t(), unacked_mutex_expire: pos_integer(), unacked_mutex_key: String.t(), virtual_qos: Kombu.Transport.Virtual.QoS.t(), visibility_timeout: pos_integer() }
Functions
Acknowledges delivery_tag, removing it from Redis unacked / unacked_index and local QoS.
@spec append(t(), Kombu.Message.t(), String.t()) :: t()
Appends a delivered message to the Redis unacked hash and unacked_index sorted set,
as well as local channel QoS state.
@spec can_consume_max_estimate(t()) :: non_neg_integer() | nil
@spec new(Kombu.Transport.Redis.PrefixedClient.t(), keyword()) :: t()
Rejects delivery_tag. If requeue: true, restores the message to the front of its queue (leftmost: true).
Removes delivery_tag from unacked_index and unacked in Redis.
@spec restore_by_tag(t(), String.t(), restore_fun(), keyword()) :: :ok
Looks up delivery_tag in unacked, deletes it from unacked and unacked_index,
and invokes restore_fun.(raw_message, exchange, routing_key, leftmost) if found.
@spec restore_unacked(t(), restore_fun()) :: t()
Restores all unacknowledged messages tracked by this channel instance.
@spec restore_unacked_once(t(), restore_fun()) :: t()
Restores unacked messages once during channel shutdown if restore_at_shutdown is enabled.
@spec restore_visible(t(), restore_fun(), keyword()) :: {:ok, non_neg_integer(), t()} | {:throttled, t()} | {:locked, t()}
Scans unacked_index for messages older than visibility_timeout and restores them,
holding a distributed Redis mutex (unacked_mutex) during restoration.
@spec set_prefetch_count(t(), non_neg_integer()) :: t()