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

Shared virtual channel operations (kombu.transport.virtual.base.Channel).

Provides message envelope preparation, base64 body encoding/decoding, priority clamping, exchange/queue/binding state transitions, and routing table lookup.

Summary

Functions

augment_message(message, exchange, routing_key, opts \\ [])

Augments a prepared message envelope with body_encoding, delivery_tag, and delivery_info (exchange, routing_key).

decode_body(body, encoding \\ "base64")

Decodes body using encoding ("base64" or "utf-8" / nil).

encode_body(body, encoding \\ "base64")

Encodes body using encoding ("base64" or "utf-8" / nil). Returns {encoded_body, encoding}.

exchange_declare(state, exchange, opts \\ [])

Declares an exchange in %BrokerState{}. Raises Kombu.ChannelError if passive: true and the exchange does not exist, or Kombu.NotEquivalentError if redeclared with incompatible attributes.

flow(active \\ true)

Raises Kombu.NotImplementedError for unsupported flow/1 on virtual channels.

get_message_priority(message, opts \\ [])

Extracts and clamps the integer priority (0..9) from message["properties"]["priority"]. If reverse: true, returns max_priority - priority (so priority 9 becomes 0 for Redis keys).

list_bindings(state)

Lists all {queue, exchange, routing_key} bindings in %BrokerState{}.

lookup(state, exchange, routing_key, default \\ nil)

Looks up all queues matching routing_key for exchange in %BrokerState{}.

message_to_python(msg, channel)

Converts a raw message map into a %Kombu.Message{} bound to channel.

prepare_message(body, opts \\ [])

Prepares a message envelope map before publishing (Channel.prepare_message).

queue_bind(state, queue, exchange, routing_key, opts \\ [])

Binds queue to exchange with routing_key in %BrokerState{}. Returns {meta_tuple, updated_broker_state}.

queue_unbind(state, queue, exchange, routing_key, opts \\ [])

Unbinds queue from exchange with routing_key in %BrokerState{}.

Functions

augment_message(message, exchange, routing_key, opts \\ [])

@spec augment_message(map(), String.t() | nil, String.t() | nil, keyword()) :: map()

Augments a prepared message envelope with body_encoding, delivery_tag, and delivery_info (exchange, routing_key).

decode_body(body, encoding \\ "base64")

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

Decodes body using encoding ("base64" or "utf-8" / nil).

encode_body(body, encoding \\ "base64")

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

Encodes body using encoding ("base64" or "utf-8" / nil). Returns {encoded_body, encoding}.

exchange_declare(state, exchange, opts \\ [])

@spec exchange_declare(
  Kombu.Transport.Virtual.BrokerState.t(),
  String.t() | nil,
  keyword()
) ::
  {:ok, Kombu.Transport.Virtual.BrokerState.t()}

Declares an exchange in %BrokerState{}. Raises Kombu.ChannelError if passive: true and the exchange does not exist, or Kombu.NotEquivalentError if redeclared with incompatible attributes.

flow(active \\ true)

Raises Kombu.NotImplementedError for unsupported flow/1 on virtual channels.

get_message_priority(message, opts \\ [])

@spec get_message_priority(map(), keyword()) :: integer()

Extracts and clamps the integer priority (0..9) from message["properties"]["priority"]. If reverse: true, returns max_priority - priority (so priority 9 becomes 0 for Redis keys).

list_bindings(state)

@spec list_bindings(Kombu.Transport.Virtual.BrokerState.t()) :: [
  {String.t(), String.t(), String.t()}
]

Lists all {queue, exchange, routing_key} bindings in %BrokerState{}.

lookup(state, exchange, routing_key, default \\ nil)

@spec lookup(
  Kombu.Transport.Virtual.BrokerState.t(),
  String.t() | nil,
  String.t() | nil,
  String.t() | nil
) :: [String.t()]

Looks up all queues matching routing_key for exchange in %BrokerState{}.

message_to_python(msg, channel)

@spec message_to_python(map() | Kombu.Message.t(), term()) :: Kombu.Message.t()

Converts a raw message map into a %Kombu.Message{} bound to channel.

prepare_message(body, opts \\ [])

@spec prepare_message(term(), keyword()) :: map()

Prepares a message envelope map before publishing (Channel.prepare_message).

queue_bind(state, queue, exchange, routing_key, opts \\ [])

@spec queue_bind(
  Kombu.Transport.Virtual.BrokerState.t(),
  String.t(),
  String.t() | nil,
  String.t() | nil,
  keyword()
) ::
  {{String.t(), String.t(), String.t()},
   Kombu.Transport.Virtual.BrokerState.t()}

Binds queue to exchange with routing_key in %BrokerState{}. Returns {meta_tuple, updated_broker_state}.

queue_unbind(state, queue, exchange, routing_key, opts \\ [])

@spec queue_unbind(
  Kombu.Transport.Virtual.BrokerState.t(),
  String.t(),
  String.t() | nil,
  String.t() | nil,
  keyword()
) :: Kombu.Transport.Virtual.BrokerState.t()

Unbinds queue from exchange with routing_key in %BrokerState{}.