Reader

Reader

A reader is actually just a consumer without a cursor. This means that Pulsar doesn't keep track of your progress and there is no need to acknowledge messages.

Creating a reader

Start with creating a client.

When creating a reader we have these options:

• ReaderName - Set the reader name. This is optional.
• MessagePrefetchCount - Number of messages that will be prefetched. The default is 1000.
• ReadCompacted - Whether to read from the compacted topic. The default is 'false'.
• StartMessageId - The initial reader position is set to the specified message-id. This is required.
• Topic - Set the topic for this reader. This is required.

Only the start message id and topic are required.

The class 'MessageId' has a public constructor and four properties: LedgerId, EntryId, Partition and BatchIndex. This enables us to save and create a MessageId and use it as the starting point for a reader (StartMessageId). If we want to start at the beginning of the topic, then use the static MessageId.Earliest as StartMessageId. If we want to start at the end of the topic, then use the static MessageId.Latest as StartMessageId.

Create a reader using the builder

var reader = client.NewReader()
                   .StartMessageId(MessageId.Earliest)
                   .Topic("persistent://public/default/mytopic")
                   .Create();

Create a reader without the builder

var options = new ReaderOptions(MessageId.Earliest, "persistent://public/default/mytopic");
var reader = client.CreateReader(options);

Receiving messages

We can either 'Receive' a single message or have an IAsyncEnumerable for 'Messages'.

Receive a single message

If we just want a single message.

var message = await reader.Receive();
Console.WriteLine("Received: " + Encoding.UTF8.GetString(message.Data.ToArray()));

Receive messages as an IAsyncEnumerable

There's an extension method in "DotPulsar.Extensions" for reading messages as an IAsyncEnumerable.

await foreach (var message in reader.Messages())
{
    Console.WriteLine("Received: " + Encoding.UTF8.GetString(message.Data.ToArray()));
}

Monitoring reader state

Monitoring the state is recommended and described here.

Disposing the reader

The Reader implements IAsyncDisposable and should be disposed when it's no longer needed. Disposing the client will dispose all readers.