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
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
Summary
Functions
Creates a stream of messages from a Pulsar topic.
Functions
@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(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 ofnon_neg_integer/0,non_neg_integer/0values) - Start from a{ledger_id, entry_id}.:start_timestamp(non_neg_integer/0) - Start from a publish time, in milliseconds since the epoch.:read_compacted(boolean/0) - Read only the latest value per key from a compacted topic. The default value isfalse.:flow_permits(pos_integer/0) - Messages to request from the broker at a time. The default value is100.:timeout(timeout/0) - Milliseconds without a message after which the stream halts. The default value is60000.:startup_timeout(timeout/0) - Milliseconds to wait for topology discovery and consumer initialization. The default value is5000.:startup_delay_ms(non_neg_integer/0) - Delay before the underlying consumer subscribes. The default value is0.:startup_jitter_ms(non_neg_integer/0) - Random delay added to:startup_delay_ms. The default value is0.
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