Kombu.Simple.SimpleQueue (Kombu v0.1.0)

Simple API for persistent queues.

Wraps a Kombu.Consumer and Kombu.Producer to offer a familiar queue-like interface with put, get, get_nowait, qsize, clear, and close.

Summary

Types

t()

Functions

clear(simple_queue)

Purges all messages from the queue.

close(simple_queue)

Closes the simple queue, cancelling consumers and stopping state agent.

empty?(sq)

Returns true if the queue has no messages.

get(sq, opts \\ [])

Retrieves a message from the queue.

get_nowait(sq)

Non-blocking retrieval of a message from the queue (Queue.get). Raises Kombu.Empty if no message is available.

new(channel, name_or_queue, opts \\ [])

Creates a new SimpleQueue.

put(simple_queue, message, opts \\ [])

Publishes a message to the queue.

qsize(simple_queue)

Returns the current number of messages in the queue.

Types

t()

@type t() :: %Kombu.Simple.SimpleQueue{
  channel: term(),
  consumer: Kombu.Consumer.t(),
  no_ack: boolean(),
  producer: Kombu.Producer.t(),
  queue: Kombu.Queue.t(),
  state_pid: pid()
}

Functions

clear(simple_queue)

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

Purges all messages from the queue.

close(simple_queue)

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

Closes the simple queue, cancelling consumers and stopping state agent.

empty?(sq)

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

Returns true if the queue has no messages.

get(sq, opts \\ [])

@spec get(t(), keyword()) :: Kombu.Message.t()

Retrieves a message from the queue.

If block: false, returns immediately or raises Kombu.Empty. If block: true, waits up to timeout seconds for a message.

get_nowait(sq)

@spec get_nowait(t()) :: Kombu.Message.t()

Non-blocking retrieval of a message from the queue (Queue.get). Raises Kombu.Empty if no message is available.

new(channel, name_or_queue, opts \\ [])

@spec new(term(), String.t() | Kombu.Queue.t(), keyword()) :: t()

Creates a new SimpleQueue.

channel can be a %Kombu.Connection{} or channel. name_or_queue can be a binary queue name or a %Kombu.Queue{} struct.

Options:

:no_ack - whether received messages are automatically acknowledged (default false)
:queue_opts - keyword options for creating the queue
:queue_args - queue arguments map
:exchange_opts - exchange options (defaults to [type: "direct"])
:serializer - default serialization format
:compression - default compression format
:accept - accepted content types list

put(simple_queue, message, opts \\ [])

@spec put(t(), term(), keyword()) :: term()

Publishes a message to the queue.

qsize(simple_queue)

@spec qsize(t()) :: non_neg_integer()

Returns the current number of messages in the queue.