# `Pulsar.Producer`
[🔗](https://github.com/efcasado/pulsar-elixir/blob/main/lib/pulsar/producer.ex#L1)

A producer publishes messages to a topic.

This module is how you add, publish through and stop producers. To declare them
on a client instead, so they start and restart with it, see `Pulsar.Client`.

`send/3` publishes, taking a producer's pid or the name it was registered under:

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

A partitioned topic needs nothing special at the call site: messages are routed across
partitions, honouring a message's `:partition_key` when one is set.

`start/1` adds a producer to a running client and `stop/2` removes it. Operations target
the logical producer by its stable root or registered name without exposing its partition
workers. `await_ready/2` waits for its topology and configured workers when an operation
must not observe asynchronous startup.

## Options

* `:topic` (`t:String.t/0`) - Required. Topic to publish to.

* `:client` (`t:atom/0`) - Client the producer belongs to. The default value is `:default`.

* `:name` - Name the producer is registered under. Defaults to `"<topic>-producer"`.

* `:access_mode` - How the topic is shared with other producers. `:shared` allows several,
  `:exclusive` fails if one is already connected, `:wait_for_exclusive` waits for
  it to disconnect, and `:exclusive_with_fencing` evicts it. The default value is `:shared`.

* `:compression` - Compression applied to the payload. The default value is `:none`.

* `:hashing_scheme` - How a message's `:partition_key` is hashed to pick a partition of a partitioned topic.

  `:murmur3_32` is implemented identically by every Pulsar client, so keys co-locate with
  producers in other languages. `:java_string_hash` matches what the Java and Go clients
  use when left at their own default.

  `:phash2_legacy` is the non-standard `:erlang.phash2/2` routing used before 3.0. No
  other client can reproduce it, so it is only for upgrading a partitioned topic without
  remapping keys mid-flight: keeping a key's existing partition preserves its ordering,
  which switching schemes breaks until the old partition drains. Prefer draining and
  moving to `:murmur3_32`.

  The default value is `:murmur3_32`.

* `:batch_enabled` (`t:boolean/0`) - Collect messages and publish them together. A message sent with `:deliver_at_time` or
  `:deliver_after` goes on its own, since the broker delays whole entries. The default value is `false`.

* `:batch_size` (`t:pos_integer/0`) - Messages to collect before flushing a batch. Only used when batching. The default value is `100`.

* `:batch_builder` - How a flushed batch is divided into entries. Only used when batching.

  `:default` publishes it as one entry, under the key of its first message. `:key_based`
  publishes one entry per key, so a `:key_shared` subscription dispatches every message on
  its own key. It regroups the batch — order holds within a key, not across them — and
  suits a small key space: `:batch_size` caps the whole batch, so mostly unique keys leave
  an entry per message and batching only adds overhead.

  The default value is `:default`.

* `:flush_interval` (`t:pos_integer/0`) - Milliseconds between batch flushes. Only used when batching. The default value is `10`.

* `:chunking_enabled` (`t:boolean/0`) - Split payloads larger than `:max_message_size` across several messages. Cannot be
  combined with `:batch_enabled`. A payload is compressed before it is measured, so
  one that compresses below the limit is sent whole. The default value is `false`.

* `:send_timeout` - Milliseconds a send may wait before its caller is answered `{:error, :send_timeout}`,
  counted from when the producer takes it rather than from when it reaches the broker.
  `false` waits indefinitely; `nil` is an alias.

  Leave it on unless something else bounds the wait: an unacknowledged send otherwise holds
  its caller, and its place in `:max_pending_messages`, until the producer restarts. Keep
  `:flush_interval` well below it.

  The default value is `30000`.

* `:max_pending_messages` - How many sends a producer will carry before refusing more with
  `{:error, :producer_queue_full}`. `false` removes the limit; `nil` is an alias.

  Counts every send taken and not yet answered, waiting to be batched or waiting for a
  receipt. A chunked message counts once, however many frames carry it, so a producer
  chunking large payloads tracks more frames than this bounds.

  The default value is `1000`.

* `:max_message_size` (`t:pos_integer/0`) - Largest chunk payload to send, in bytes. Only used when chunking. Capped by the limit
  the broker advertises when the producer connects, minus the metadata each chunk carries. The default value is `5242880`.

* `:schema` (`t:keyword/0`) - Schema to register with the topic, as `[type: atom, definition: term]`. See
  `Pulsar.Schema`.

* `:partition_discovery_interval_ms` - For a partitioned topic, how often to look for partitions added since startup.
  `false` disables later metadata checks, but not initial topic discovery or local
  recovery of groups that have stopped. The default value is `60000`.

* `:startup_delay_ms` (`t:non_neg_integer/0`) - Delay before a producer connects. A broker that is not connected yet is retried, so this is only needed to stagger a large number of restarts. The default value is `0`.

* `:startup_jitter_ms` (`t:non_neg_integer/0`) - Random extra delay on top of `:startup_delay_ms`, to spread out restarts. The default value is `0`.

# `chunked_message_id`

```elixir
@type chunked_message_id() :: %{
  first_chunk_message_id: Pulsar.Protocol.Binary.Pulsar.Proto.MessageIdData.t(),
  last_chunk_message_id: Pulsar.Protocol.Binary.Pulsar.Proto.MessageIdData.t(),
  uuid: String.t(),
  num_chunks: pos_integer()
}
```

What a chunked send is answered with, since its message spans several broker messages.

# `send_result`

```elixir
@type send_result() ::
  Pulsar.Protocol.Binary.Pulsar.Proto.MessageIdData.t()
  | chunked_message_id()
  | :deduplicated
```

What `send/3` is answered with. `:deduplicated` carries no message id because the broker
assigned none; see `send/3`.

# `await`

```elixir
@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.

# `await_ready`

```elixir
@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`

# `send`

```elixir
@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 `t: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")

# `send_async`

```elixir
@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`.

# `start`

```elixir
@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.

# `start`

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

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

# `start_link`

```elixir
@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.

# `stop`

```elixir
@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.

---

*Consult [api-reference.md](api-reference.md) for complete listing*