Kombu.Transport.Virtual.QoS (Kombu v0.1.0)

Virtual channel Quality of Service (kombu.transport.virtual.base.QoS). Manages prefetch_count limits and unacknowledged message tracking/restoration.

Summary

Types

delivery_tag()

t()

Functions

ack(qos, delivery_tag)

Marks delivery_tag as acknowledged (dirty).

active_count(qo_s)

Returns the number of currently active (delivered minus dirty) messages.

append(qos, message, delivery_tag)

Appends a message with delivery_tag to the unacknowledged state (flushing dirty tags first).

can_consume?(qos)

Returns true if the channel can consume more messages given prefetch_count.

can_consume_max_estimate(qos)

Returns the estimated maximum number of messages that can be consumed at once, or nil when prefetch_count == 0.

clear(qos)

Clears all delivered and dirty tags from %QoS{}.

flush(qos)

Flushes dirty (acked/rejected) delivery tags from delivered.

get(qo_s, delivery_tag)

Gets a delivered message by delivery_tag. Raises KeyError if not found.

new(prefetch_count \\ 0, opts \\ [])

Creates a new %QoS{} state.

reject(qos, delivery_tag, opts \\ [])

Marks delivery_tag as rejected, optionally invoking restore_fun.(message) when requeue: true.

restore_unacked(qos, restore_fun)

Restores all unacknowledged messages in LIFO order using restore_fun.(message). Returns {errors, updated_qos}.

restore_unacked_once(qos, restore_fun, opts \\ [])

Restores all unacknowledged messages at most once on channel close/shutdown.

Types

delivery_tag()

@type delivery_tag() :: String.t() | integer()

t()

@type t() :: %Kombu.Transport.Virtual.QoS{
  delivered_map: %{required(delivery_tag()) => term()},
  delivered_order: [delivery_tag()],
  dirty: MapSet.t(delivery_tag()),
  prefetch_count: non_neg_integer(),
  restore_at_shutdown: boolean(),
  restored: boolean()
}

Functions

ack(qos, delivery_tag)

@spec ack(t(), delivery_tag()) :: t()

Marks delivery_tag as acknowledged (dirty).

active_count(qo_s)

@spec active_count(t()) :: non_neg_integer()

Returns the number of currently active (delivered minus dirty) messages.

append(qos, message, delivery_tag)

@spec append(t(), term(), delivery_tag()) :: t()

Appends a message with delivery_tag to the unacknowledged state (flushing dirty tags first).

can_consume?(qos)

@spec can_consume?(t()) :: boolean()

Returns true if the channel can consume more messages given prefetch_count.

can_consume_max_estimate(qos)

@spec can_consume_max_estimate(t()) :: non_neg_integer() | nil

Returns the estimated maximum number of messages that can be consumed at once, or nil when prefetch_count == 0.

clear(qos)

@spec clear(t()) :: t()

Clears all delivered and dirty tags from %QoS{}.

flush(qos)

@spec flush(t()) :: t()

Flushes dirty (acked/rejected) delivery tags from delivered.

get(qo_s, delivery_tag)

@spec get(t(), delivery_tag()) :: term()

Gets a delivered message by delivery_tag. Raises KeyError if not found.

new(prefetch_count \\ 0, opts \\ [])

@spec new(non_neg_integer(), keyword()) :: t()

Creates a new %QoS{} state.

reject(qos, delivery_tag, opts \\ [])

@spec reject(t(), delivery_tag(), keyword()) :: t()

Marks delivery_tag as rejected, optionally invoking restore_fun.(message) when requeue: true.

restore_unacked(qos, restore_fun)

@spec restore_unacked(t(), (term() -> any())) :: {[{Exception.t(), term()}], t()}

Restores all unacknowledged messages in LIFO order using restore_fun.(message). Returns {errors, updated_qos}.

restore_unacked_once(qos, restore_fun, opts \\ [])

@spec restore_unacked_once(t(), (term() -> any()), keyword()) ::
  {[{Exception.t(), term()}], t()}

Restores all unacknowledged messages at most once on channel close/shutdown.