Kombu.Connection (Kombu v0.1.0)

Copy Markdown View Source

Broker connection (kombu.Connection), managing transport resolution, URL parsing, multi-URL failover (round-robin / shuffle), lazy connection & default channel creation, and operation retry (ensure_connection/2, ensure/4, autoretry/3).

Summary

Types

t()

Functions

add_declared_entity(connection, entity_key)

Records entity_key in the connection's declared_entities set.

as_uri(conn, opts \\ [])

Formats the connection parameters as a URL string (Connection.as_uri).

autoretry(conn, fun, opts \\ [])

Wraps fun (which accepts channel as its last argument or via :channel option) with automatic reconnection and channel revival, returning {result, channel}.

channel(conn)

Creates and returns a new channel process for conn.

channel_pool(conn, limit \\ nil, opts \\ [])
channel_promise(conn)

Returns a 0-arity function that lazily returns the connection's default_channel (Connection.channel_promise).

check_url_transport!(name)

Validates a transport scheme extracted from a URL (forbidding . or :).

clear_declared_entities(connection)

Clears all declared entities on conn.

clone(conn, overrides \\ [])

Creates a clone of conn with optional parameter overrides.

close(conn)

Alias for release/1.

collect(conn, socket_timeout \\ nil)

Cleans up connection resources and clears declared_entities.

completes_cycle?(connection, retries)

Returns true if a failover cycle round is complete after retries (0-indexed).

connect(conn)

Establishes the connection immediately (max_retries: 1). Returns the underlying transport connection map.

connected?(connection)

Returns true if the connection is currently established and viable.

connection(conn)

Returns the underlying transport connection map, connecting lazily if needed.

consumer(conn, queues \\ [], opts \\ [])
declared_entities(connection)

Returns the set of entities declared on this connection session.

default_channel(conn)

Returns the default channel for conn, creating it lazily if needed.

drain_events(conn, opts \\ [])

Waits for a single event from the broker on conn's default channel.

ensure(conn, obj, fun, opts \\ [])

Wraps an operation fun on obj so that recoverable connection/channel errors trigger reconnection, revive, and retry (Connection.ensure).

ensure_connection(conn, opts \\ [])

Ensures conn has an active connection to the broker, retrying according to opts and rotating through alternate URLs if configured. Returns conn.

get_transport_cls(transport)

Resolves a transport identifier ("redis", "rediss", "memory", or module atom) to its transport module.

has_declared_entity?(conn, entity_key)

Returns true if entity_key has been declared on this connection session.

heartbeat_check(connection, opts \\ [])

Checks heartbeats (no-op for transports that do not use heartbeats).

host(conn)

Returns "hostname:port" for the connection.

info(conn)

Returns connection parameters as a map (Connection.info).

is_connection?(arg1)

Returns true if obj is a %Kombu.Connection{}.

maybe_channel(conn)

Resolves a connection, channel promise, or channel into a concrete channel.

maybe_close_channel(arg1, channel)

Closes channel, ignoring any connection or channel errors.

maybe_switch_next(conn)

Switches to the next alternate URL according to the configured failover strategy.

new(url_or_opts \\ "redis://localhost:6379/0", opts \\ [])

Creates a new %Kombu.Connection{} from a URL string, list of URLs, or keyword options.

pool(conn, limit \\ nil, opts \\ [])
producer(conn, opts \\ [])
producer_pool(conn, limit \\ nil, opts \\ [])
recoverable_channel_errors(connection)

Returns the list of recoverable channel error modules for conn.

recoverable_connection_errors(connection)

Returns the list of recoverable connection error modules for conn.

release(conn)

Closes the connection and its default channel, clearing declared entities.

reset_declared_entities(conn)

Alias for clear_declared_entities/1.

revive(conn, new_channel)

Revives the connection after re-establishment, closing old default channel if replaced.

simple_buffer(conn, name, opts \\ [])
simple_queue(conn, name, opts \\ [])
supports_exchange_type?(conn, exchange_type)

Returns true if the connection's transport supports exchange_type.

switch(conn, conn_str)

Switches connection parameters to conn_str (URL or hostname) without reconnecting.

Types

t()

@type t() :: %Kombu.Connection{
  alternates: term(),
  connect_timeout: term(),
  credential_provider: term(),
  extra_opts: term(),
  failover_strategy: term(),
  heartbeat: term(),
  hostname: term(),
  initial_params: term(),
  insist: term(),
  login_method: term(),
  password: term(),
  port: term(),
  ssl: term(),
  state_pid: term(),
  transport: term(),
  transport_options: term(),
  uri_prefix: term(),
  userid: term(),
  virtual_host: term()
}

Functions

add_declared_entity(connection, entity_key)

@spec add_declared_entity(t(), term()) :: :ok

Records entity_key in the connection's declared_entities set.

as_uri(conn, opts \\ [])

@spec as_uri(t(), keyword()) :: String.t()

Formats the connection parameters as a URL string (Connection.as_uri).

autoretry(conn, fun, opts \\ [])

@spec autoretry(t(), function(), keyword()) :: function()

Wraps fun (which accepts channel as its last argument or via :channel option) with automatic reconnection and channel revival, returning {result, channel}.

channel(conn)

@spec channel(t()) :: pid()

Creates and returns a new channel process for conn.

channel_pool(conn, limit \\ nil, opts \\ [])

channel_promise(conn)

@spec channel_promise(t()) :: (-> pid())

Returns a 0-arity function that lazily returns the connection's default_channel (Connection.channel_promise).

check_url_transport!(name)

@spec check_url_transport!(String.t()) :: String.t()

Validates a transport scheme extracted from a URL (forbidding . or :).

clear_declared_entities(connection)

@spec clear_declared_entities(t()) :: :ok

Clears all declared entities on conn.

clone(conn, overrides \\ [])

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

Creates a clone of conn with optional parameter overrides.

close(conn)

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

Alias for release/1.

collect(conn, socket_timeout \\ nil)

@spec collect(t(), number() | nil) :: t()

Cleans up connection resources and clears declared_entities.

completes_cycle?(connection, retries)

@spec completes_cycle?(t(), non_neg_integer()) :: boolean()

Returns true if a failover cycle round is complete after retries (0-indexed).

connect(conn)

@spec connect(t()) :: map()

Establishes the connection immediately (max_retries: 1). Returns the underlying transport connection map.

connected?(connection)

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

Returns true if the connection is currently established and viable.

connection(conn)

@spec connection(t()) :: map() | nil

Returns the underlying transport connection map, connecting lazily if needed.

consumer(conn, queues \\ [], opts \\ [])

declared_entities(connection)

@spec declared_entities(t()) :: MapSet.t()

Returns the set of entities declared on this connection session.

default_channel(conn)

@spec default_channel(t()) :: pid()

Returns the default channel for conn, creating it lazily if needed.

drain_events(conn, opts \\ [])

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

Waits for a single event from the broker on conn's default channel.

ensure(conn, obj, fun, opts \\ [])

@spec ensure(t(), term(), function(), keyword()) :: function()

Wraps an operation fun on obj so that recoverable connection/channel errors trigger reconnection, revive, and retry (Connection.ensure).

Returns a function accepting either () (0-arity) or (args_list) (1-arity) that invokes fun with automatic retry and revival.

ensure_connection(conn, opts \\ [])

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

Ensures conn has an active connection to the broker, retrying according to opts and rotating through alternate URLs if configured. Returns conn.

get_transport_cls(transport)

@spec get_transport_cls(String.t() | atom()) :: module()

Resolves a transport identifier ("redis", "rediss", "memory", or module atom) to its transport module.

has_declared_entity?(conn, entity_key)

@spec has_declared_entity?(t(), term()) :: boolean()

Returns true if entity_key has been declared on this connection session.

heartbeat_check(connection, opts \\ [])

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

Checks heartbeats (no-op for transports that do not use heartbeats).

host(conn)

@spec host(t()) :: String.t()

Returns "hostname:port" for the connection.

info(conn)

@spec info(t()) :: map()

Returns connection parameters as a map (Connection.info).

is_connection?(arg1)

@spec is_connection?(term()) :: boolean()

Returns true if obj is a %Kombu.Connection{}.

maybe_channel(conn)

@spec maybe_channel(t() | function() | Kombu.Channel.t()) :: Kombu.Channel.t()

Resolves a connection, channel promise, or channel into a concrete channel.

maybe_close_channel(arg1, channel)

@spec maybe_close_channel(t(), pid() | nil) :: :ok

Closes channel, ignoring any connection or channel errors.

maybe_switch_next(conn)

@spec maybe_switch_next(t()) :: t()

Switches to the next alternate URL according to the configured failover strategy.

new(url_or_opts \\ "redis://localhost:6379/0", opts \\ [])

@spec new(String.t() | [String.t()] | keyword(), keyword()) :: t()

Creates a new %Kombu.Connection{} from a URL string, list of URLs, or keyword options.

pool(conn, limit \\ nil, opts \\ [])

producer(conn, opts \\ [])

producer_pool(conn, limit \\ nil, opts \\ [])

recoverable_channel_errors(connection)

Returns the list of recoverable channel error modules for conn.

recoverable_connection_errors(connection)

Returns the list of recoverable connection error modules for conn.

release(conn)

@spec release(t()) :: t()

Closes the connection and its default channel, clearing declared entities.

reset_declared_entities(conn)

@spec reset_declared_entities(t()) :: :ok

Alias for clear_declared_entities/1.

revive(conn, new_channel)

@spec revive(t(), pid() | nil) :: t()

Revives the connection after re-establishment, closing old default channel if replaced.

simple_buffer(conn, name, opts \\ [])

simple_queue(conn, name, opts \\ [])

supports_exchange_type?(conn, exchange_type)

@spec supports_exchange_type?(t(), String.t()) :: boolean()

Returns true if the connection's transport supports exchange_type.

switch(conn, conn_str)

@spec switch(t(), String.t()) :: t()

Switches connection parameters to conn_str (URL or hostname) without reconnecting.