Kombu.Message (Kombu v0.1.0)

Base class for received messages (kombu.message.Message and kombu.transport.virtual.base.Message).

Summary

Types

t()

Functions

ack(msg, opts \\ [])

Acknowledges the message (Message.ack) and returns the updated %Kombu.Message{state: "ACK"}. Raises Kombu.MessageStateError if the message is not bound to a channel or is already acknowledged.

ack_log_error(msg, error_modules \\ [Exception], opts \\ [])

Acknowledges the message, logging any errors matching errors instead of raising.

acknowledged?(message)

Returns true if the message has been acknowledged, rejected, or requeued.

decode(msg)

Deserializes the message body according to its content_type, content_encoding, and accept list.

decode_body(body, arg2)

Decodes a message body according to body_encoding ("base64" or nil / "string").

decode_cached(msg)

Deserializes the message body and caches the result on the returned struct.

encode_body(body, arg2)

Encodes a message body according to body_encoding ("base64" or nil / "string").

new(opts_or_raw \\ [], channel_or_opts \\ nil)

Creates a new %Kombu.Message{} from keyword options or from a raw virtual transport message map (new(raw_message_map, channel)).

payload(msg)

Returns the deserialized message payload (Message.payload).

reject(msg, opts \\ [])

Rejects the message (Message.reject) and returns %Kombu.Message{state: "REJECTED"}.

reject_log_error(msg, error_modules \\ [Exception], opts \\ [])

Rejects the message, logging any errors matching errors instead of raising.

requeue(msg)

Rejects and requeues the message (Message.requeue), returning %Kombu.Message{state: "REQUEUED"}.

reraise_error!(msg, callback \\ nil)

Reraises the first error encountered during message initialization (such as decompression failure), or invokes callback.(err) if provided.

serializable(msg)

Converts a virtual-transport message back into its serializable envelope map (kombu.transport.virtual.base.Message.serializable).

Types

t()

@type t() :: %Kombu.Message{
  accept: [String.t()] | MapSet.t() | nil,
  body: binary() | term() | nil,
  channel: Kombu.Channel.t() | nil,
  content_encoding: String.t() | nil,
  content_type: String.t() | nil,
  decoded_cache: {:cached, term()} | nil,
  delivery_info: map(),
  delivery_tag: String.t() | integer() | nil,
  errors: [Exception.t()],
  headers: map(),
  postencode: String.t() | nil,
  properties: map(),
  raw_message: map() | nil,
  state: String.t()
}

Functions

ack(msg, opts \\ [])

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

Acknowledges the message (Message.ack) and returns the updated %Kombu.Message{state: "ACK"}. Raises Kombu.MessageStateError if the message is not bound to a channel or is already acknowledged.

ack_log_error(msg, error_modules \\ [Exception], opts \\ [])

@spec ack_log_error(t(), [module()], keyword()) :: t()

Acknowledges the message, logging any errors matching errors instead of raising.

acknowledged?(message)

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

Returns true if the message has been acknowledged, rejected, or requeued.

decode(msg)

@spec decode(t()) :: term()

Deserializes the message body according to its content_type, content_encoding, and accept list.

decode_body(body, arg2)

@spec decode_body(term(), String.t() | nil) :: term()

Decodes a message body according to body_encoding ("base64" or nil / "string").

decode_cached(msg)

@spec decode_cached(t()) :: {term(), t()}

Deserializes the message body and caches the result on the returned struct.

encode_body(body, arg2)

@spec encode_body(term(), String.t() | nil) :: term()

Encodes a message body according to body_encoding ("base64" or nil / "string").

new(opts_or_raw \\ [], channel_or_opts \\ nil)

@spec new(keyword() | map(), Kombu.Channel.t() | keyword() | nil) :: t()

Creates a new %Kombu.Message{} from keyword options or from a raw virtual transport message map (new(raw_message_map, channel)).

payload(msg)

@spec payload(t()) :: term()

Returns the deserialized message payload (Message.payload).

reject(msg, opts \\ [])

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

Rejects the message (Message.reject) and returns %Kombu.Message{state: "REJECTED"}.

reject_log_error(msg, error_modules \\ [Exception], opts \\ [])

@spec reject_log_error(t(), [module()], keyword()) :: t()

Rejects the message, logging any errors matching errors instead of raising.

requeue(msg)

@spec requeue(t()) :: t()

Rejects and requeues the message (Message.requeue), returning %Kombu.Message{state: "REQUEUED"}.

reraise_error!(msg, callback \\ nil)

@spec reraise_error!(t(), (Exception.t() -> any()) | nil) :: :ok

Reraises the first error encountered during message initialization (such as decompression failure), or invokes callback.(err) if provided.

serializable(msg)

@spec serializable(t()) :: map()

Converts a virtual-transport message back into its serializable envelope map (kombu.transport.virtual.base.Message.serializable).