Kombu.Consumer (Kombu v0.1.0)

Message Consumer (kombu.Consumer).

Manages queue declarations, QoS prefetch settings, consumer tags, content-type acceptance checks (DisallowedSerializerError), automatic message decoding, on_decode_error handling, and callback dispatch.

Summary

Types

t()

Functions

accept(consumer)

Returns the accepted content types for this consumer.

active_tags(consumer)

Returns the active consumer tags map.

add_queue(consumer, queue)

Adds a queue to the consumer without starting consumption (Consumer.add_queue). Call consume/2 afterwards to begin consuming from the added queue.

cancel(consumer)

Cancels all active queue consumers (Consumer.cancel / Consumer.close).

cancel_by_queue(consumer, queue_or_name)

Cancels consumption from a single queue and removes it from the consumer (Consumer.cancel_by_queue).

channel(consumer)

Returns the channel or channel promise for this consumer.

close(consumer)

consume(consumer, opts \\ [])

Starts consuming messages from all registered queues (Consumer.consume).

consuming_from?(consumer, queue_or_name)

Returns true if consumer is currently consuming from queue_or_name (Consumer.consuming_from).

declare(consumer)

Declares all registered queues, exchanges, and bindings (Consumer.declare).

flow(consumer, active)

Enables/disables flow control on the consumer's channel (Consumer.flow).

new(channel_or_conn, queues \\ [], opts \\ [])

Creates a new %Kombu.Consumer{}.

purge(consumer)

Purges all registered queues and returns the total number of purged messages (Consumer.purge).

qos(consumer, opts \\ [])

Specifies quality of service (prefetch_size, prefetch_count, apply_global) on the consumer's channel (Consumer.qos).

queues(consumer)

Returns the list of queues currently tracked by the consumer.

receive_message(consumer, body, message)

Dispatches a decoded message (body, message) to all registered callbacks (Consumer.receive).

recover(consumer, opts \\ [])

Redelivers unacknowledged messages on the consumer's channel (Consumer.recover).

register_callback(consumer, callback)

Registers a new callback fn body, message -> ... end (Consumer.register_callback).

revive(consumer, channel_or_conn)

Revives the consumer on a new channel or connection (Consumer.revive).

with_consumer(consumer, fun)

Executes fun.(consumer) while the consumer is active, automatically cancelling all queue consumers when fun completes (with Consumer(...) as consumer:).

Types

t()

@type t() :: %Kombu.Consumer{
  accept: term(),
  active_queue_no_ack: term(),
  active_tags: term(),
  auto_declare: term(),
  callbacks: term(),
  channel: term(),
  connection: term(),
  no_ack: term(),
  on_decode_error: term(),
  on_message: term(),
  prefetch_count: term(),
  queues: term(),
  state_pid: term(),
  tag_prefix: term()
}

Functions

accept(consumer)

@spec accept(t()) :: [String.t()] | nil

Returns the accepted content types for this consumer.

active_tags(consumer)

@spec active_tags(t()) :: map()

Returns the active consumer tags map.

add_queue(consumer, queue)

@spec add_queue(t(), Kombu.Queue.t()) :: {Kombu.Queue.t(), t()}

Adds a queue to the consumer without starting consumption (Consumer.add_queue). Call consume/2 afterwards to begin consuming from the added queue.

cancel(consumer)

@spec cancel(t()) :: t()

Cancels all active queue consumers (Consumer.cancel / Consumer.close).

cancel_by_queue(consumer, queue_or_name)

@spec cancel_by_queue(t(), Kombu.Queue.t() | String.t()) :: t()

Cancels consumption from a single queue and removes it from the consumer (Consumer.cancel_by_queue).

channel(consumer)

@spec channel(t()) :: Kombu.Channel.t() | function() | nil

Returns the channel or channel promise for this consumer.

close(consumer)

@spec close(t()) :: t()

consume(consumer, opts \\ [])

@spec consume(t(), keyword()) :: t()

Starts consuming messages from all registered queues (Consumer.consume).

consuming_from?(consumer, queue_or_name)

@spec consuming_from?(t(), Kombu.Queue.t() | String.t()) :: boolean()

Returns true if consumer is currently consuming from queue_or_name (Consumer.consuming_from).

declare(consumer)

@spec declare(t()) :: t()

Declares all registered queues, exchanges, and bindings (Consumer.declare).

flow(consumer, active)

@spec flow(t(), boolean()) :: term()

Enables/disables flow control on the consumer's channel (Consumer.flow).

new(channel_or_conn, queues \\ [], opts \\ [])

@spec new(
  Kombu.Connection.t() | Kombu.Channel.t() | nil,
  Kombu.Queue.t() | [Kombu.Queue.t()] | nil,
  keyword()
) :: t()

Creates a new %Kombu.Consumer{}.

purge(consumer)

@spec purge(t()) :: non_neg_integer()

Purges all registered queues and returns the total number of purged messages (Consumer.purge).

qos(consumer, opts \\ [])

@spec qos(t(), keyword()) :: :ok

Specifies quality of service (prefetch_size, prefetch_count, apply_global) on the consumer's channel (Consumer.qos).

queues(consumer)

@spec queues(t()) :: [Kombu.Queue.t()]

Returns the list of queues currently tracked by the consumer.

receive_message(consumer, body, message)

@spec receive_message(t(), term(), Kombu.Message.t()) :: :ok

Dispatches a decoded message (body, message) to all registered callbacks (Consumer.receive).

recover(consumer, opts \\ [])

@spec recover(t(), keyword()) :: :ok

Redelivers unacknowledged messages on the consumer's channel (Consumer.recover).

register_callback(consumer, callback)

@spec register_callback(t(), (term(), Kombu.Message.t() -> term())) :: t()

Registers a new callback fn body, message -> ... end (Consumer.register_callback).

revive(consumer, channel_or_conn)

@spec revive(t(), Kombu.Connection.t() | Kombu.Channel.t()) :: t()

Revives the consumer on a new channel or connection (Consumer.revive).

with_consumer(consumer, fun)

@spec with_consumer(t(), (t() -> result)) :: result when result: var

Executes fun.(consumer) while the consumer is active, automatically cancelling all queue consumers when fun completes (with Consumer(...) as consumer:).