Middlewares not fully implemented

src/KafkaFlow.Abstractions/Configuration/IClusterConfigurationBuilder.cs

@@ -85,7 +85,7 @@ IClusterConfigurationBuilder CreateTopicIfNotExists(
 
         // Gets the cluster configuration
         IClusterConfigurationBuilder AddInstrumentation<TConsumerInstrumentationMiddleware, TProducerInstrumentationMiddleware>()
-            where TConsumerInstrumentationMiddleware : class, IMessageMiddleware
-            where TProducerInstrumentationMiddleware : class, IMessageMiddleware;
+            where TConsumerInstrumentationMiddleware : class, IConsumerInstrumentationMiddleware
+            where TProducerInstrumentationMiddleware : class, IProducerInstrumentationMiddleware;
     }
 }

src/KafkaFlow.Abstractions/IConsumerContext.cs

@@ -1,6 +1,7 @@
 namespace KafkaFlow
 {
     using System;
+    using System.Collections.Generic;
     using System.Threading;
 
     /// <summary>
@@ -73,5 +74,7 @@ public interface IConsumerContext
         /// Resume Kafka's message fetch
         /// </summary>
         void Resume();
+
+        IReadOnlyCollection<string> Brokers { get; }
     }
 }

src/KafkaFlow.Abstractions/IConsumerInstrumentationMiddleware.cs

@@ -0,0 +1,6 @@
+namespace KafkaFlow
+{
+    public interface IConsumerInstrumentationMiddleware : IMessageMiddleware
+    {
+    }
+}

src/KafkaFlow.Abstractions/IProducerInstrumentationMiddleware.cs

@@ -0,0 +1,6 @@
+namespace KafkaFlow
+{
+    public interface IProducerInstrumentationMiddleware : IMessageMiddleware
+    {
+    }
+}

src/KafkaFlow.OpenTelemetry/Trace/TracerConsumerMiddleware.cs

@@ -8,7 +8,7 @@
     using global::OpenTelemetry;
     using global::OpenTelemetry.Context.Propagation;
 
-    internal class TracerConsumerMiddleware : IMessageMiddleware
+    internal class TracerConsumerMiddleware : IConsumerInstrumentationMiddleware
     {
         private static readonly TextMapPropagator Propagator = Propagators.DefaultTextMapPropagator;
         private static readonly string ProcessString = "process";

src/KafkaFlow.OpenTelemetry/Trace/TracerProducerMiddleware.cs

@@ -8,7 +8,7 @@
     using global::OpenTelemetry;
     using global::OpenTelemetry.Context.Propagation;
 
-    internal class TracerProducerMiddleware : IMessageMiddleware
+    internal class TracerProducerMiddleware : IProducerInstrumentationMiddleware
     {
         private static readonly TextMapPropagator Propagator = Propagators.DefaultTextMapPropagator;
         private static readonly string PublishString = "publish";

src/KafkaFlow/Configuration/ClusterConfigurationBuilder.cs

@@ -39,9 +39,9 @@ public ClusterConfiguration Build(KafkaConfiguration kafkaConfiguration)
             configuration.AddProducers(this.producers.Select(x => x.Build(configuration)));
             configuration.AddConsumers(this.consumers.Select(x =>
             {
-                if(this.instrumentationConsumerMiddleware != null)
+                if (this.instrumentationConsumerMiddleware != null)
                 {
-                    x.AddInstrumentation<this.instrumentationConsumerMiddleware>();
+                    x.AddInstrumentation<IMessageMiddleware>();
                 }
 
                 return x.Build(configuration);
@@ -107,11 +107,15 @@ public IClusterConfigurationBuilder AddConsumer(Action<IConsumerConfigurationBui
         }
 
         public IClusterConfigurationBuilder AddInstrumentation<TConsumerInstrumentationMiddleware, TProducerInstrumentationMiddleware>()
-            where TConsumerInstrumentationMiddleware : class, IMessageMiddleware
-            where TProducerInstrumentationMiddleware : class, IMessageMiddleware
+            where TConsumerInstrumentationMiddleware : class, IConsumerInstrumentationMiddleware
+            where TProducerInstrumentationMiddleware : class, IProducerInstrumentationMiddleware
         {
-
             this.instrumentationConsumerMiddleware = typeof(TConsumerInstrumentationMiddleware);
+            this.instrumentationProducerMiddleware = typeof(IProducerInstrumentationMiddleware);
+
+            //this.DependencyConfigurator
+            //    .AddTransient(typeof(IConsumerInstrumentationMiddleware), this.instrumentationConsumerMiddleware)
+            //    .AddTransient(typeof(IProducerInstrumentationMiddleware), this.instrumentationProducerMiddleware);
 
             return this;
         }

src/KafkaFlow/Configuration/ConsumerConfigurationBuilder.cs

@@ -229,10 +229,10 @@ public IConsumerConfigurationBuilder WithCustomFactory(ConsumerCustomFactory cus
             return this;
         }
 
-        public void AddInstrumentation<T>()
+        public void AddInstrumentation<T>(IConsumerInstrumentationMiddleware consumerInstrumentationMiddleware)
             where T : class, IMessageMiddleware
         {
-            this.middlewareConfigurationBuilder.AddAtBeginning<T>();
+            var type = typeof(IConsumerInstrumentationMiddleware);            
         }
 
         public IConsumerConfiguration Build(ClusterConfiguration clusterConfiguration)

src/KafkaFlow/Configuration/MiddlewareConfigurationBuilder.cs

@@ -44,6 +44,11 @@ public TBuilder AddAtBeginning<T>(MiddlewareLifetime lifetime = MiddlewareLifeti
             return this.AddAt<T>(0, lifetime);
         }
 
+        public TBuilder AddAtBeginning(Type messageMiddleware, MiddlewareLifetime lifetime = MiddlewareLifetime.ConsumerOrProducer)
+        {
+            return this.AddAt(messageMiddleware, 0, lifetime);
+        }
+
         public IReadOnlyList<MiddlewareConfiguration> Build() => this.middlewaresConfigurations;
 
         private static InstanceLifetime ParseLifetime(MiddlewareLifetime lifetime)
@@ -92,5 +97,20 @@ private TBuilder AddAt<T>(
 
             return this as TBuilder;
         }
+
+        private TBuilder AddAt(
+            Type messageMiddleware,
+            int position,
+            MiddlewareLifetime lifetime = MiddlewareLifetime.ConsumerOrProducer)
+        {
+
+            this.DependencyConfigurator.Add(typeof(IMessageMiddleware), messageMiddleware, ParseLifetime(lifetime));
+
+            this.middlewaresConfigurations.Insert(
+                position,
+                new MiddlewareConfiguration(messageMiddleware.GetType(), lifetime));
+
+            return this as TBuilder;
+        }
     }
 }

src/KafkaFlow/Consumers/ConsumerContext.cs

@@ -1,6 +1,7 @@
 namespace KafkaFlow.Consumers
 {
     using System;
+    using System.Collections.Generic;
     using System.Threading;
     using Confluent.Kafka;
 
@@ -38,6 +39,8 @@ public ConsumerContext(
 
         public string GroupId => this.consumer.Configuration.GroupId;
 
+        public IReadOnlyCollection<string> Brokers => this.consumer.Configuration.ClusterConfiguration.Brokers;
+
         public bool ShouldStoreOffset { get; set; } = true;
 
         public DateTime MessageTimestamp => this.kafkaResult.Message.Timestamp.UtcDateTime;