@spec await(reference(), timeout()) :: {:ok, send_result()} | {:error, term()}

Waits for a send started by send_async/3, answering as send/3 would have.

A producer that goes down before answering is reported as {:error, {:producer_died, reason}}.

Each reference can be awaited once. The default timeout is :infinity; when the producer's :send_timeout is disabled, this can wait indefinitely. A finite timeout abandons the wait without cancelling the send, so the message may still be published. It also consumes the reference: a late answer is dropped and cannot be recovered by calling await/2 again.

@spec await_ready(
  pid() | String.t() | atom(),
  keyword()
) :: :ok | {:error, :not_found | :timeout}

Waits for a producer and all its configured workers to be ready.

Takes the stable root returned by start/1 or its registered name. A named producer is resolved repeatedly, so the wait also tolerates its client or resource branch restarting.

Readiness means initial topic discovery and topology construction have completed, and every configured worker has registered with its broker. A worker that repeatedly fails registration causes the wait to time out. Readiness is a snapshot: it does not guarantee continued broker availability or prevent a worker from restarting immediately afterward.

Options:

• :timeout - maximum time to wait in milliseconds, or :infinity; defaults to 5 seconds
• :client - client name or pid used to resolve a producer name; defaults to :default

@spec send(pid() | String.t() | atom(), binary(), keyword()) ::
  {:ok, send_result()} | {:error, term()}

Publishes a message, given a producer's pid or name.

Returns {:error, :not_ready} while its topic topology is being discovered.

A message too large for the broker is refused here rather than sent, since the broker would answer it by closing a connection shared with every other producer and consumer:

• {:error, :message_too_large} for a message, or a whole batch, that does not fit. With :chunking_enabled the payload is split to fit, so this only surfaces if the broker's limit is not yet known to the producer
• {:error, :metadata_too_large} with :chunking_enabled, when :properties and the rest of the metadata leave no room for a payload to be split into

A producer already carrying :max_pending_messages refuses with {:error, :producer_queue_full} rather than taking on more.

Calling send/3 from the selected producer worker returns {:error, :calling_self}.

{:error, :send_timeout} is the other shape: the broker did not acknowledge the message within :send_timeout. It does not say the message was not published, only that nothing came back in time. A retry publishes under a fresh sequence id, which the broker's deduplication does not match against the first attempt, so it can duplicate the message.

A successful send answers with the broker's message id, or with a chunked_message_id/0 when :chunking_enabled split the payload.

With :batch_enabled it answers when the broker acknowledges the entry the message was batched into, not when the message joins the batch, so the wait includes up to :flush_interval before anything is sent at all.

On a topic with deduplication enabled it can instead answer {:ok, :deduplicated}: the broker recognised the sequence id as one it had already stored, kept the message it had, and assigned this call no message id. Deduplication matches on the sequence id alone and never on the payload, so the message it kept is not necessarily the one passed here. Before 3.0 this was reported as {:ok, message_id}, with a message id referring to nothing.

Options

• :partition_key - decides the partition of a partitioned topic, hashed under the producer's :hashing_scheme, and is carried with the message so a :key_shared subscription can use it. Must be a binary; before 3.0 any term was accepted here
• :properties - a map of user properties carried with the message
• :event_time - the message's event time, in milliseconds
• :deliver_at_time / :deliver_after - delayed delivery. The broker delays whole entries, so a delayed message is published on its own rather than joining a batch
• :timeout - how long to wait, in milliseconds, answering {:error, :timeout} if it passes. Defaults to :infinity. With the default producer settings, :send_timeout remains the deadline; if :send_timeout is disabled, the wait can be unbounded. :send_timeout is counted from when the producer takes the message, so it does not bound the wait for a producer that has not finished registering. Giving up here does not cancel the send
• :client - the client to resolve a producer name against

Examples

{:ok, message_id} = Pulsar.Producer.send(:audit, "payload")
{:ok, message_id} = Pulsar.Producer.send(:audit, "payload", partition_key: "tenant-1")

@spec send_async(pid() | String.t() | atom(), binary(), keyword()) ::
  {:ok, reference()} | {:error, term()}

Starts publishing without waiting for the broker and returns a reference for await/2.

Takes the same options as send/3, except :timeout, which belongs to await/2. Calls from one process that route to the same partition are handed to the producer in order.

{:ok, first} = Pulsar.Producer.send_async(:audit, "one")
{:ok, second} = Pulsar.Producer.send_async(:audit, "two")

{:ok, _message_id} = Pulsar.Producer.await(first)
{:ok, _message_id} = Pulsar.Producer.await(second)

The process that calls send_async/3 must also call await/2, because the reply and the producer's :DOWN message arrive in its mailbox. If the reference is never awaited, the reply remains unread there.

Routing failures, such as a producer still discovering its topology, are returned immediately. Once a worker accepts the send, producer errors such as a full queue are returned by await/2.

@spec start(keyword() | String.t()) :: DynamicSupervisor.on_start_child()

Adds a producer to a running client.

For producers whose set is only known at runtime. Prefer the client's :producers for ones known up front: a producer added here is not recreated if the client restarts.

Returns once the stable producer supervisor has been registered. Topic discovery and worker initialization continue asynchronously; publishing returns {:error, :not_ready} until discovery completes.

@spec start(
  String.t(),
  keyword()
) :: DynamicSupervisor.on_start_child()

Same as start/1, with the topic given positionally.

@spec start_link(keyword()) :: Supervisor.on_start()

Starts a producer, linked to the calling process.

Returns the stable producer root. See the module documentation for the options.

@spec stop(
  pid() | String.t() | atom(),
  keyword()
) :: :ok | {:error, :not_found}

Stops a producer, given its pid or its name.

A pid must be the stable root returned by start/1 or start_link/1. Group and worker pids are not producer roots and return {:error, :not_found} here.

A root started as a static child will be restarted by its supervisor; remove that child from the supervision tree instead.