From be1569e7dc80556de7b716b323d8fba644389604 Mon Sep 17 00:00:00 2001
From: Ivan Argenio <ivan.argenio@repower.com>
Date: Fri, 13 Oct 2023 15:12:18 +0200
Subject: [PATCH] Added producerOnly param to EnableAdminMessages method

---
 .../ClusterConfigurationBuilderExtensions.cs  | 79 +++++++++++--------
 1 file changed, 45 insertions(+), 34 deletions(-)

diff --git a/src/KafkaFlow.Admin/Extensions/ClusterConfigurationBuilderExtensions.cs b/src/KafkaFlow.Admin/Extensions/ClusterConfigurationBuilderExtensions.cs
index 4e8c00477..e710b6263 100644
--- a/src/KafkaFlow.Admin/Extensions/ClusterConfigurationBuilderExtensions.cs
+++ b/src/KafkaFlow.Admin/Extensions/ClusterConfigurationBuilderExtensions.cs
@@ -20,12 +20,14 @@ public static class ClusterConfigurationBuilderExtensions
         /// <param name="topic">The topic to be used by the admin commands</param>
         /// <param name="consumerGroup">The consumer group prefix</param>
         /// <param name="topicPartition">The partition used to produce and consumer admin messages</param>
+        /// <param name="producerOnly">To produce admin messages only</param>
         /// <returns></returns>
         public static IClusterConfigurationBuilder EnableAdminMessages(
             this IClusterConfigurationBuilder cluster,
             string topic,
             string consumerGroup = null,
-            int topicPartition = 0)
+            int topicPartition = 0,
+            bool producerOnly = false)
         {
             consumerGroup ??= $"Admin-{Assembly.GetEntryAssembly()!.GetName().Name}";
 
@@ -33,45 +35,54 @@ public static IClusterConfigurationBuilder EnableAdminMessages(
                 .AddSingleton<IAdminProducer>(
                     resolver => new AdminProducer(
                         resolver.Resolve<IMessageProducer<AdminProducer>>(),
-                        topicPartition))
-                .AddSingleton<IConsumerAdmin, ConsumerAdmin>();
+                        topicPartition));
 
-            return cluster
+            cluster
                 .AddProducer<AdminProducer>(
                     producer => producer
                         .DefaultTopic(topic)
                         .AddMiddlewares(
                             middlewares => middlewares
-                                .AddSerializer<ProtobufNetSerializer>()))
-                .AddConsumer(
-                    consumer => consumer
-                        .ManualAssignPartitions(topic, new[] { topicPartition })
-                        .WithGroupId(consumerGroup)
-                        .WithoutStoringOffsets()
-                        .WithWorkersCount(1)
-                        .WithBufferSize(1)
-                        .WithAutoOffsetReset(AutoOffsetReset.Latest)
-                        .DisableManagement()
-                        .AddMiddlewares(
-                            middlewares => middlewares
-                                .AddSerializer<ProtobufNetSerializer>()
-                                .AddTypedHandlers(
-                                    handlers => handlers
-                                        .WithHandlerLifetime(InstanceLifetime.Singleton)
-                                        .AddHandlers(
-                                            new[]
-                                            {
-                                                typeof(ChangeConsumerWorkersCountHandler),
-                                                typeof(PauseConsumerByNameHandler),
-                                                typeof(PauseConsumersByGroupHandler),
-                                                typeof(ResetConsumerOffsetHandler),
-                                                typeof(RestartConsumerByNameHandler),
-                                                typeof(ResumeConsumerByNameHandler),
-                                                typeof(ResumeConsumersByGroupHandler),
-                                                typeof(RewindConsumerOffsetToDateTimeHandler),
-                                                typeof(StartConsumerByNameHandler),
-                                                typeof(StopConsumerByNameHandler),
-                                            }))));
+                                .AddSerializer<ProtobufNetSerializer>()));
+
+            if (!producerOnly)
+            {
+                cluster.DependencyConfigurator
+                    .AddSingleton<IConsumerAdmin, ConsumerAdmin>();
+
+                cluster
+                    .AddConsumer(
+                        consumer => consumer
+                            .ManualAssignPartitions(topic, new[] { topicPartition })
+                            .WithGroupId(consumerGroup)
+                            .WithoutStoringOffsets()
+                            .WithWorkersCount(1)
+                            .WithBufferSize(1)
+                            .WithAutoOffsetReset(AutoOffsetReset.Latest)
+                            .DisableManagement()
+                            .AddMiddlewares(
+                                middlewares => middlewares
+                                    .AddSerializer<ProtobufNetSerializer>()
+                                    .AddTypedHandlers(
+                                        handlers => handlers
+                                            .WithHandlerLifetime(InstanceLifetime.Singleton)
+                                            .AddHandlers(
+                                                new[]
+                                                {
+                                                    typeof(ChangeConsumerWorkersCountHandler),
+                                                    typeof(PauseConsumerByNameHandler),
+                                                    typeof(PauseConsumersByGroupHandler),
+                                                    typeof(ResetConsumerOffsetHandler),
+                                                    typeof(RestartConsumerByNameHandler),
+                                                    typeof(ResumeConsumerByNameHandler),
+                                                    typeof(ResumeConsumersByGroupHandler),
+                                                    typeof(RewindConsumerOffsetToDateTimeHandler),
+                                                    typeof(StartConsumerByNameHandler),
+                                                    typeof(StopConsumerByNameHandler),
+                                                }))));
+            }
+
+            return cluster;
         }
 
         /// <summary>