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

A high-level interface for reading messages from Pulsar topics using
Elixir's Stream API. The reader uses non-durable subscriptions, meaning
it doesn't persist its position and starts fresh on each connection.

## Usage

A reader reads through a client, so there has to be one running. In an application it
belongs in your supervision tree; in a script, start it directly:

    {:ok, _pid} = Pulsar.Client.start_link(host: "pulsar://localhost:6650")

    Pulsar.Reader.stream("persistent://public/default/my-topic", start_position: :earliest)
    |> Stream.take(10)
    |> Stream.each(&IO.inspect(&1.payload))
    |> Stream.run()

    :ok = Pulsar.Client.stop(:default)

Reading always uses a non-durable subscription, so a reader keeps no position and
starts fresh each time.

Selecting a client, when there is more than the default:

    Pulsar.Reader.stream(topic, client: :analytics)
    |> Stream.map(&process/1)
    |> Stream.run()

With custom flow control:

    Pulsar.Reader.stream(topic, flow_permits: 50) |> Enum.take(100)

Reading from a specific message:

    # {ledger_id, entry_id}
    Pulsar.Reader.stream(topic, start_message_id: {123, 456}) |> Enum.take(10)

## Options

See `stream/2`.

## Partitioned Topics

The Reader supports partitioned topics. When reading from a partitioned topic,
messages from all partitions are merged into a single stream. **Note that message
ordering across partitions is not guaranteed** - messages may arrive interleaved
from different partitions.

If you need per-partition ordering, consider using separate Reader streams for
each partition (e.g., `"persistent://tenant/ns/topic-partition-0"`).

## Process Ownership

The stream is bound to the process that creates it. Messages are delivered to
the creating process's mailbox, so you cannot pass the stream to another process
for consumption.

For multi-process consumption patterns, use the `Pulsar.Consumer` API directly
or consider [off_broadway_pulsar](https://github.com/efcasado/off_broadway_pulsar)
for Broadway-based pipelines.

## Stream Termination

The stream terminates when any of these conditions is met:
- The consumer receives all requested messages (e.g., via `Enum.take/2`)
- The inactivity timeout is reached (default: 60 seconds)
- The stream is halted by downstream processing

# `stream`

```elixir
@spec stream(
  String.t(),
  keyword()
) :: Enumerable.t()
```

Creates a stream of messages from a Pulsar topic.

Returns a `Stream` that yields `Pulsar.Message` structs. If initialization
fails, the stream emits `{:error, reason}` as the first (and only) element.

## Options

* `:client` (`t:atom/0`) - The client to read through. The default value is `:default`.

* `:start_position` - Where to start reading when no message id or timestamp is given. The default value is `:earliest`.

* `:start_message_id` (tuple of `t:non_neg_integer/0`, `t:non_neg_integer/0` values) - Start from a `{ledger_id, entry_id}`.

* `:start_timestamp` (`t:non_neg_integer/0`) - Start from a publish time, in milliseconds since the epoch.

* `:read_compacted` (`t:boolean/0`) - Read only the latest value per key from a compacted topic. The default value is `false`.

* `:flow_permits` (`t:pos_integer/0`) - Messages to request from the broker at a time. The default value is `100`.

* `:timeout` (`t:timeout/0`) - Milliseconds without a message after which the stream halts. The default value is `60000`.

* `:startup_timeout` (`t:timeout/0`) - Milliseconds to wait for topology discovery and consumer initialization. The default value is `5000`.

* `:startup_delay_ms` (`t:non_neg_integer/0`) - Delay before the underlying consumer subscribes. The default value is `0`.

* `:startup_jitter_ms` (`t:non_neg_integer/0`) - Random delay added to `:startup_delay_ms`. The default value is `0`.

## Examples

    Pulsar.Reader.stream("persistent://public/default/topic", start_position: :earliest)
    |> Enum.take(5)

    Pulsar.Reader.stream("persistent://public/default/topic",
      client: :analytics,
      start_position: :latest
    )
    |> Stream.filter(&interesting?/1)
    |> Enum.to_list()

    # Handle errors (emitted as first element if initialization fails)
    Pulsar.Reader.stream("persistent://public/default/topic", client: :not_running)
    |> Enum.take(1)
    |> case do
      [{:error, reason}] -> Logger.error("Failed: #{inspect(reason)}")
      messages -> process(messages)
    end

---

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