automated-testing-demo/consumer/Consumer.cs

53 lines
1.5 KiB
C#

using System;
using System.Threading;
using Confluent.Kafka;
using Model;
using Serializers;
class Consumer
{
public static void Main(string[] args)
{
var conf = new ConsumerConfig{
GroupId = "test-consumer-group",
BootstrapServers = "localhost:29092",
AutoOffsetReset = AutoOffsetReset.Latest
};
using (var consumer = new ConsumerBuilder<Ignore, DataModel>(conf)
.SetValueDeserializer(new JsonSerializer<DataModel>())
.Build())
{
consumer.Subscribe("test-topic");
CancellationTokenSource cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => {
e.Cancel = true;
cts.Cancel();
};
try
{
while(true)
{
try
{
var cr = consumer.Consume(cts.Token);
Console.WriteLine($"Consumed message with id '{cr.Message.Value.id}' at: '{cr.TopicPartitionOffset}'.");
}
catch (ConsumeException e)
{
Console.WriteLine($"Error occured: {e.Error.Reason}");
}
}
}
catch (OperationCanceledException)
{
// Ensure the consumer leaves the group cleanly and final offsets are committed.
consumer.Close();
}
}
}
}