ViennaNET/src/ViennaNET.Messaging.KafkaQueue/KafkaConnectionFactory.cs at master · Raiffeisen-DGTL/ViennaNET · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
using System;
using Confluent.Kafka;
using ViennaNET.Utils;

namespace ViennaNET.Messaging.KafkaQueue
{
  internal class KafkaConnectionFactory : IKafkaConnectionFactory
  {
    public IConsumer<byte[], byte[]> CreateConsumer(
      KafkaQueueConfiguration config,
      Action<IConsumer<byte[], byte[]>, LogMessage>? logHandler = null,
      Action<IConsumer<byte[], byte[]>, Error>? errorHandler = null)
    {
      config.ThrowIfNull(nameof(config));
      config.ConsumerConfig.ThrowIfNull(nameof(config.ProducerConfig));

      config.ConsumerConfig!.EnableAutoCommit ??= !config.TransactionEnabled;
      config.ConsumerConfig.IsolationLevel ??= config.TransactionEnabled
        ? IsolationLevel.ReadCommitted
        : IsolationLevel.ReadUncommitted;

      var builder = new ConsumerBuilder<byte[], byte[]>(config.ConsumerConfig);

      if (logHandler != null)
      {
        builder.SetLogHandler(logHandler);
      }

      if (errorHandler != null)
      {
        builder.SetErrorHandler(errorHandler);
      }

      return builder.Build();
    }

    public IProducer<byte[], byte[]> CreateProducer(
      KafkaQueueConfiguration config,
      Action<IProducer<byte[], byte[]>, LogMessage>? logHandler = null,
      Action<IProducer<byte[], byte[]>, Error>? errorHandler = null)
    {
      config.ThrowIfNull(nameof(config));
      config.ProducerConfig.ThrowIfNull(nameof(config.ProducerConfig));

      config.ProducerConfig!.TransactionalId = config.TransactionEnabled
        ? config.Id
        : null;

      var builder = new ProducerBuilder<byte[], byte[]>(config.ProducerConfig);

      if (logHandler != null)
      {
        builder.SetLogHandler(logHandler);
      }

      if (errorHandler != null)
      {
        builder.SetErrorHandler(errorHandler);
      }

      return builder.Build();
    }
  }
}