using System;
using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using xEventService.Configuration;
using xEventService.Constants;
namespace xEventService.Providers
{
///
/// an Abstract Kafka Consumer Service ...
///
public abstract class XKafkaConsumerServiceBase : BackgroundService
{
///
/// Consumed Topic Messages ...
///
private readonly string topic;
///
/// Kafka Consumer Class ...
///
public readonly IConsumer consumer;
private readonly XEventServiceConfiguration configuration;
private readonly ILogger logger;
protected XKafkaConsumerServiceBase(
XEventServiceConfiguration configuration,
ILogger logger,
string topic = XEventServiceConstants.XDefaultTopic
)
{
//
this.topic = topic;
this.logger = logger;
this.configuration = configuration;
//
// Prepare Kafka Consumer Configuration ...
var config = new ConsumerConfig
{
EnableAutoCommit = false,
BootstrapServers = configuration.Url,
AutoOffsetReset = AutoOffsetReset.Earliest,
GroupId = XEventServiceConstants.XDefaultConsumerGroup,
};
consumer = new ConsumerBuilder(config).Build();
}
///
/// Dispose Implementation ...
///
public override void Dispose()
{
//
consumer.Close();
consumer.Dispose();
GC.SuppressFinalize(this);
base.Dispose();
}
}
}