Home - News ( United States | United Kingdom | Italy | Germany ) - Football scores

Showing content from https://github.com/apache/pulsar-dotpulsar/wiki/Consumer below:

Consumer · apache/pulsar-dotpulsar Wiki · GitHub

Need to get messages from Pulsar and acknowledge them, so that Pulsar can keep track of your progress? Then a consumer is what we want.

Start with creating a client.

When creating a consumer we have these options:

• ConsumerName - Set the consumer name. This is optional.
• InitialPosition - Set initial position for the subscription. The default is 'Latest'.
• PriorityLevel - Set the priority level for the shared subscription consumer. The default is 0.
• MessagePrefetchCount - Number of messages that will be prefetched. The default is 1000.
• ReadCompacted - Whether to read from the compacted topic. The default is 'false'.
• SubscriptionName - Set the subscription name for this consumer. This is required.
• SubscriptionType - Set the subscription type for this consumer. The default is 'Exclusive'.
• Topic - Set the topic for this consumer. This is required.

Only the subscription name and topic are required.

var consumer = client.NewConsumer()
                     .SubscriptionName("MySubscription")
                     .Topic("persistent://public/default/mytopic")
                     .Create();

var options = new ConsumerOptions("MySubscription", "persistent://public/default/mytopic");
var consumer = client.CreateConsumer(options);

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

If we just want a single message.

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

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

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

The message has these properties:

• MessageId - The id of the message.
• Data - The data payload as a ReadOnlySequence.
• EventTime - Set the event time of the message.
• Key - Set the key of the message for routing policy.
• KeyBytes - Set the key of the message for routing policy.
• OrderingKey - Set the ordering key of the message for message dispatch in SubscriptionType.KeyShared mode.
• ProducerName - The name of the producer that produced the message.
• PublishTime - When the message was published.
• SequenceId - Set the sequence id of the message.

We have two options when acknowledging messages.

To just acknowledge a single message:

await consumer.Acknowledge(message);

To acknowledge all messages up to and include then provided message:

await consumer.AcknowledgeCumulative(message);

Please do note: We can't do this if the subscription is shared!

Need to reset the cursor? What's easy:

await consumer.Seek(MessageId.Earliest); // Provide the message-id to reset to.

Please do note: When seeking we will shortly be disconnected, as that's how Pulsar handles seek requests.

As easy as:

await consumer.Unsubscribe();

Please do note: We can't use the consumer after unsubscribing and therefore should dispose it.

Monitoring the state is recommended and described here.

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

RetroSearch is an open source project built by @garambo | Open a GitHub Issue

Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo

HTML: 3.2 | Encoding: UTF-8 | Version: 0.7.4