Resolve Comments

Source/BSN.Commons/Infrastructure/Kafka/IKafkaConsumer.cs

@@ -8,7 +8,7 @@ namespace BSN.Commons.Infrastructure.Kafka
     /// Represents A Kafka Api compatible consumer class
     /// </summary>
     /// <typeparam name="T">The object type that is being consumed</typeparam>
-    public interface IKafkaConsumer<T> : IDisposable
+    public interface IKafkaConsumer<T>
     {
         /// <summary>
         /// Consumes the Kafka api 

Source/BSN.Commons/Infrastructure/Kafka/IKafkaProducer.cs

@@ -20,7 +20,7 @@ public KafkaProduceException(string message) : base(message) {}
     /// Represents A Kafka Api compatible producer class
     /// </summary>
     /// <typeparam name="T">The object type that is being produced</typeparam>
-    public interface IKafkaProducer<T> : IDisposable
+    public interface IKafkaProducer<T>
     {
         /// <summary>
         /// Produces the Kafka api, raises exceptions for errors

Source/BSN.Commons/Infrastructure/Kafka/KafkaConsumer.cs

@@ -25,11 +25,5 @@ public Task<T> ConsumeAsync(CancellationToken ct = default)
         }
 
         private readonly IConsumer<Null, T> _consumer;
-
-        /// <inheritdoc />
-        public void Dispose()
-        {
-            _consumer?.Dispose();
-        }
     }
 }

Source/BSN.Commons/Infrastructure/Kafka/KafkaConsumerFactory.cs

@@ -1,4 +1,6 @@
 using Confluent.Kafka;
+using Microsoft.Extensions.Options;
+using System.Collections.Generic;
 
 namespace BSN.Commons.Infrastructure.Kafka
 {
@@ -9,36 +11,48 @@ public class KafkaConsumerFactory<T> : IKafkaConsumerFactory<T>
         /// <param name="options">Default Options for KafkaConsumers</param>
         public KafkaConsumerFactory(IKafkaConsumerOptions options)
         {
-            _defaultConsumerConfig = new ConsumerConfig
-            {
-                BootstrapServers = options.BootstrapServers,
-                AutoOffsetReset = AutoOffsetReset.Earliest
-            };
-
-            // Here we did this because the ReceiveMessageMaxBytes in ProducerConfig type
-            // is int and can not accept high values that we expect
-            _defaultConsumerConfig.Set("receive.message.max.bytes", options.ReceiveMessageMaxBytes);
+            _defaultConsumerOptions = options;
         }
 
         /// <inheritdoc />
         public IKafkaConsumer<T> Create(string topic, string groupId)
         {
-            lock (this)
+            var consumerKey = topic + ":" + groupId;
+
+            if (_consumers.ContainsKey(consumerKey))
             {
-                var config = new ConsumerConfig(_defaultConsumerConfig) { GroupId = groupId };
-                var consumer = new ConsumerBuilder<Null, T>(config).Build();
-                consumer.Subscribe(topic);
-
-                return new KafkaConsumer<T>(consumer);
+                return new KafkaConsumer<T>(_consumers[consumerKey]);
             }
+
+            var config = new ConsumerConfig() 
+            {
+                BootstrapServers = _defaultConsumerOptions.BootstrapServers,
+                AutoOffsetReset = AutoOffsetReset.Earliest,
+            };
+
+
+            // Here we did this because the ReceiveMessageMaxBytes in ProducerConfig type
+            // is int and can not accept high values that we expect
+            config.Set("receive.message.max.bytes", _defaultConsumerOptions.ReceiveMessageMaxBytes);
+
+            var consumer = new ConsumerBuilder<Null, T>(config).Build();
+            consumer.Subscribe(topic);
+
+            _consumers.Add(consumerKey, consumer);
+
+            return new KafkaConsumer<T>(consumer);
         }
 
-        private readonly ConsumerConfig _defaultConsumerConfig;
+        private readonly Dictionary<string, IConsumer<Null, T>> _consumers;
+        private IKafkaConsumerOptions _defaultConsumerOptions;
 
         /// <inheritdoc />
         public void Dispose()
         {
-            // TODO release managed resources here
+            foreach (var consumer in _consumers)
+            {
+                consumer.Value.Dispose();
+            }
         }
     }
 }

Source/BSN.Commons/Infrastructure/Kafka/KafkaConsumerOptions.cs

@@ -3,10 +3,21 @@
     /// <inheritdoc />
     public class KafkaConsumerOptions : IKafkaConsumerOptions
     {
+        /// <summary>
+        /// Initializes a new instance of the <see cref="KafkaConsumerOptions"/> class.
+        /// </summary>
+        /// <param name="bootstrapServers">List of kafka bootstrap servers seperated by ;</param>
+        /// <param name="receiveMessageMaxBytes">Represents the Size that a Receiving message can have in Bytes</param>
+        public KafkaConsumerOptions(string bootstrapServers, string receiveMessageMaxBytes)
+        {
+            BootstrapServers = bootstrapServers;
+            ReceiveMessageMaxBytes = receiveMessageMaxBytes;
+        }
+
         /// <inheritdoc />
-        public string BootstrapServers { get; set; }
+        public string BootstrapServers { get; }
 
         /// <inheritdoc />
-        public string ReceiveMessageMaxBytes { get; set; }
+        public string ReceiveMessageMaxBytes { get; }
     }
 }

Source/BSN.Commons/Infrastructure/MessageBroker/Kafka/KafkaEventAggregator.cs

@@ -43,13 +43,10 @@ ILogger logger
                 {
                     BootstrapServers = _kafkaConnectionOptions.BootstrapServers
                 });
-                
+
             _consumerFactory = new KafkaConsumerFactory<string>(
-                new KafkaConsumerOptions
-                {
-                    BootstrapServers = _kafkaConnectionOptions.BootstrapServers,
-                    ReceiveMessageMaxBytes = _kafkaConnectionOptions.ReceiveMessageMaxBytes,
-                });
+                    new KafkaConsumerOptions(bootstrapServers: _kafkaConnectionOptions.BootstrapServers, receiveMessageMaxBytes: _kafkaConnectionOptions.ReceiveMessageMaxBytes)
+                );
         }
 
         /// <summary>
@@ -151,27 +148,12 @@ public void UnSubscribe<TEvent, TEventDataModel>(IEventReceiver eventReceiver) w
 
         /// <inheritdoc />
         public void Dispose()
-        {
-            foreach (var producer in _producers)
-            {   
-                producer.Value.Dispose();
-            }
-
-            foreach (var consumer in _consumers)
+        {   
+            foreach (var ct in _consumersCancellationTokenSources)
             {
-                _consumersCancellationTokenSources[consumer.Key].Cancel();
-
-                consumer.Value.Dispose();
+                ct.Value.Cancel();
             }
 
-            _producers.Clear();
-
-            _consumers.Clear();
-
-            _consumersCancellationTokenSources.Clear();
-
-            _subscriptionManager.Clear();
-
             _producerFactory.Dispose();
 
             _consumerFactory.Dispose();