Kombu.Producer (Kombu v0.1.0)

Message Producer (kombu.Producer).

Handles serialization (Kombu.Serialization), compression (Kombu.Compression), automatic exchange/entity declaration (maybe_declare), connection retry (ensure), and batched Redis publishing (Producer.batch/3).

Summary

Types

t()

Functions

batch(producer, opts_or_max_size \\ [], fun)

Executes fun within a batched publish session (with producer.batch(max_size=...)).

channel(producer)

Resolves and returns the concrete channel for producer.

close(producer)

declare(producer)

Declares the producer's default exchange if it has a non-empty name (Producer.declare).

flush_active_batch!()

Flushes any messages currently buffered in the active Producer.batch context.

maybe_declare(producer, entity, opts \\ [])

Declares entity unless already declared during this connection session (Producer.maybe_declare).

new(channel_or_conn \\ nil, opts \\ [])

Creates a new %Kombu.Producer{}.

prepare(producer, body, opts \\ [])

Serializes and optionally compresses body for publishing (Producer._prepare).

publish(producer, body, opts \\ [])

Publishes a message to the broker (Producer.publish).

release(producer)

Releases the producer (Producer.release / Producer.close).

revive(producer, conn)

Revives the producer with a new channel or connection after connection loss (Producer.revive).

supports_batch_publish?(producer)

Returns true if the active transport supports batched publishing (Producer.supports_batch_publish).

Types

t()

@type t() :: %Kombu.Producer{
  auto_declare: term(),
  channel: term(),
  compression: term(),
  connection: term(),
  exchange: term(),
  on_return: term(),
  routing_key: term(),
  serializer: term()
}

Functions

batch(producer, opts_or_max_size \\ [], fun)

@spec batch(t(), pos_integer() | keyword(), (map() -> result) | (-> result)) :: result
when result: var

Executes fun within a batched publish session (with producer.batch(max_size=...)).

Supports nested Producer.batch blocks on the same process. When the outermost batch block exits normally, any buffered messages are flushed in a single pipeline. If an exception is raised, buffered messages are discarded.

channel(producer)

@spec channel(t()) :: Kombu.Channel.t()

Resolves and returns the concrete channel for producer.

close(producer)

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

declare(producer)

@spec declare(t()) :: :ok

Declares the producer's default exchange if it has a non-empty name (Producer.declare).

flush_active_batch!()

@spec flush_active_batch!() :: :ok

Flushes any messages currently buffered in the active Producer.batch context.

maybe_declare(producer, entity, opts \\ [])

@spec maybe_declare(t(), term(), keyword()) :: boolean()

Declares entity unless already declared during this connection session (Producer.maybe_declare).

new(channel_or_conn \\ nil, opts \\ [])

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

Creates a new %Kombu.Producer{}.

prepare(producer, body, opts \\ [])

@spec prepare(t(), term(), keyword()) :: {binary(), String.t(), String.t(), map()}

Serializes and optionally compresses body for publishing (Producer._prepare).

Returns {prepared_body, content_type, content_encoding, updated_headers}.

publish(producer, body, opts \\ [])

@spec publish(t(), term(), keyword()) :: :ok

Publishes a message to the broker (Producer.publish).

release(producer)

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

Releases the producer (Producer.release / Producer.close).

revive(producer, conn)

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

Revives the producer with a new channel or connection after connection loss (Producer.revive).

supports_batch_publish?(producer)

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

Returns true if the active transport supports batched publishing (Producer.supports_batch_publish).