# `Kombu.Consumer`
[🔗](https://github.com/hacktobeer/kombu/blob/v0.1.0/lib/kombu/consumer.ex#L1)

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.

# `t`

```elixir
@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()
}
```

# `accept`

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

Returns the accepted content types for this consumer.

# `active_tags`

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

Returns the active consumer tags map.

# `add_queue`

```elixir
@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`

```elixir
@spec cancel(t()) :: t()
```

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

# `cancel_by_queue`

```elixir
@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`

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

Returns the channel or channel promise for this consumer.

# `close`

```elixir
@spec close(t()) :: t()
```

# `consume`

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

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

# `consuming_from?`

```elixir
@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`

```elixir
@spec declare(t()) :: t()
```

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

# `flow`

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

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

# `new`

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

Creates a new `%Kombu.Consumer{}`.

# `purge`

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

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

# `qos`

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

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

# `queues`

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

Returns the list of queues currently tracked by the consumer.

# `receive_message`

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

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

# `recover`

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

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

# `register_callback`

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

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

# `revive`

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

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

# `with_consumer`

```elixir
@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:`).

---

*Consult [api-reference.md](api-reference.md) for complete listing*