Remove need of producer registration

YarinOmesi · Aug 7, 2023 · 5a2c0cb · 5a2c0cb
1 parent a7ed395
commit 5a2c0cb
Show file tree

Hide file tree

Showing 6 changed files with 36 additions and 23 deletions.
diff --git a/doc/content/3.documentation/5.configuration/2.transports/10.kafka.md b/doc/content/3.documentation/5.configuration/2.transports/10.kafka.md
@@ -236,7 +236,7 @@ public class Program
 
 #### Producer Provider
 
-With MassTransit v8.1, you can dynamically resolve a producer using `ITopicProducerProvider` (registered as `Scoped`). However, please note that producers' registration is still required.
+With MassTransit v8.1, you can dynamically resolve a producer using `ITopicProducerProvider` (registered as `Scoped`).
 
 ```csharp
 namespace KafkaProducer;
@@ -259,8 +259,6 @@ public class Program
 
             x.AddRider(rider =>
             {
-                rider.AddProducer<KafkaMessage>("topic-name");
-
                 rider.UsingKafka((context, k) => { k.Host("localhost:9092"); });
             });
         });

diff --git a/src/Transports/MassTransit.KafkaIntegration/Configuration/IKafkaHostConfiguration.cs b/src/Transports/MassTransit.KafkaIntegration/Configuration/IKafkaHostConfiguration.cs
@@ -18,6 +18,13 @@ public interface IKafkaHostConfiguration :
         KafkaSendTransportContext<TKey, TValue> CreateSendTransportContext<TKey, TValue>(IBusInstance busInstance, string topic)
             where TValue : class;
 
+        IKafkaProducerSpecification CreateSpecification<TKey, TValue>(string topicName, Action<IKafkaProducerConfigurator<TKey, TValue>> configure)
+            where TValue : class;
+
+        IKafkaProducerSpecification CreateSpecification<TKey, TValue>(string topicName, ProducerConfig producerConfig,
+            Action<IKafkaProducerConfigurator<TKey, TValue>> configure)
+            where TValue : class;
+
         IKafkaConsumerSpecification CreateSpecification<TKey, TValue>(string topicName, string groupId,
             Action<IKafkaTopicReceiveEndpointConfigurator<TKey, TValue>> configure)
             where TValue : class;

diff --git a/...afkaIntegration/ITopicProducerProvider.cs → ...afkaIntegration/ITopicProducerProvider.cs b/...afkaIntegration/ITopicProducerProvider.cs → ...afkaIntegration/ITopicProducerProvider.cs
@@ -1,4 +1,4 @@
-namespace MassTransit.KafkaIntegration
+namespace MassTransit
 {
     using System;
 

diff --git a/...s/MassTransit.KafkaIntegration/KafkaIntegration/Configuration/KafkaFactoryConfigurator.cs b/...s/MassTransit.KafkaIntegration/KafkaIntegration/Configuration/KafkaFactoryConfigurator.cs
@@ -106,17 +106,7 @@ void IKafkaFactoryConfigurator.TopicProducer<TKey, TValue>(string topicName, Pro
             if (producerConfig == null)
                 throw new ArgumentNullException(nameof(producerConfig));
 
-            var added = _producers.TryAdd(topicName, topic =>
-            {
-                var configurator = new KafkaProducerSpecification<TKey, TValue>(this, producerConfig, topicName, _oAuthBearerTokenRefreshHandler);
-                configurator.SetHeadersSerializer(_headersSerializer);
-                configure?.Invoke(configurator);
-
-                configurator.ConnectSendObserver(_sendObservers);
-                if (_configureSend != null)
-                    configurator.ConfigureSend(_configureSend);
-                return configurator;
-            });
+            var added = _producers.TryAdd(topicName, topic => CreateSpecification(topic, producerConfig, configure));
 
             if (!added)
                 throw new ConfigurationException($"A topic producer with the same key was already added: {topicName}");
@@ -271,14 +261,35 @@ public KafkaSendTransportContext<TKey, TValue> CreateSendTransportContext<TKey,
             where TValue : class
         {
             if (!_producers.TryGetValue(topic, out var spec))
-                throw new ConfigurationException($"Producer for topic: {topic} is not configured.");
+                spec = CreateSpecification<TKey, TValue>(topic);
 
             if (spec is IKafkaProducerSpecification<TKey, TValue> specification)
                 return specification.CreateSendTransportContext(busInstance);
 
             throw new ConfigurationException($"Producer for topic: {topic} is not configured for ${typeof(Message<TKey, TValue>).Name} message");
         }
 
+        public IKafkaProducerSpecification CreateSpecification<TKey, TValue>(string topicName,
+            Action<IKafkaProducerConfigurator<TKey, TValue>> configure = null)
+            where TValue : class
+        {
+            return CreateSpecification(topicName, new ProducerConfig(), configure);
+        }
+
+        public IKafkaProducerSpecification CreateSpecification<TKey, TValue>(string topicName, ProducerConfig producerConfig,
+            Action<IKafkaProducerConfigurator<TKey, TValue>> configure = null)
+            where TValue : class
+        {
+            var configurator = new KafkaProducerSpecification<TKey, TValue>(this, producerConfig, topicName, _oAuthBearerTokenRefreshHandler);
+            configurator.SetHeadersSerializer(_headersSerializer);
+            configure?.Invoke(configurator);
+
+            configurator.ConnectSendObserver(_sendObservers);
+            if (_configureSend != null)
+                configurator.ConfigureSend(_configureSend);
+            return configurator;
+        }
+
         public IKafkaConsumerSpecification CreateSpecification<TKey, TValue>(string topicName, string groupId,
             Action<IKafkaTopicReceiveEndpointConfigurator<TKey, TValue>> configure)
             where TValue : class

diff --git a/...ration/TopicProducerProviderExtensions.cs → ...n/KafkaTopicProducerProviderExtensions.cs b/...ration/TopicProducerProviderExtensions.cs → ...n/KafkaTopicProducerProviderExtensions.cs
@@ -1,12 +1,13 @@
-namespace MassTransit.KafkaIntegration
+namespace MassTransit
 {
     using System;
     using Confluent.Kafka;
     using DependencyInjection;
+    using KafkaIntegration;
     using Microsoft.Extensions.DependencyInjection;
 
 
-    public static class TopicProducerProviderExtensions
+    public static class KafkaTopicProducerProviderExtensions
     {
         public static ITopicProducer<TValue> GetProducer<TValue>(this ITopicProducerProvider provider, Uri address)
             where TValue : class
@@ -22,7 +23,7 @@ public static ITopicProducer<TValue> GetProducer<TKey, TValue>(this ITopicProduc
             return new KeyedTopicProducer<TKey, TValue>(producer, keyResolver);
         }
 
-        public static ITopicProducerProvider GetScopedTopicProducerProvider(this ITopicProducerProvider producerProvider, IServiceProvider provider)
+        internal static ITopicProducerProvider GetScopedTopicProducerProvider(this ITopicProducerProvider producerProvider, IServiceProvider provider)
         {
             var contextProvider = provider.GetService<IScopedConsumeContextProvider>();
             return contextProvider is { HasContext: true }

diff --git a/tests/MassTransit.KafkaIntegration.Tests/Producer_Specs.cs b/tests/MassTransit.KafkaIntegration.Tests/Producer_Specs.cs
@@ -34,7 +34,6 @@ public async Task Should_receive_messages()
                         r.AddConsumer<TestKafkaMessageConsumer<KafkaMessage>>();
 
                         r.AddProducer<KafkaMessage>(Topic, producerConfig);
-
                         r.UsingKafka((context, k) =>
                         {
                             k.TopicEndpoint<KafkaMessage>(Topic, consumerConfig, c =>
@@ -110,7 +109,6 @@ public class Producer_Provider_Specs
         public async Task Should_receive_messages()
         {
             var consumerConfig = new ConsumerConfig { GroupId = nameof(Producer_Provider_Specs) };
-            var producerConfig = new ProducerConfig();
 
             await using var provider = new ServiceCollection()
                 .ConfigureKafkaTestOptions(options =>
@@ -125,8 +123,6 @@ public async Task Should_receive_messages()
                     x.SetTestTimeouts(testInactivityTimeout: TimeSpan.FromSeconds(15));
                     x.AddRider(r =>
                     {
-                        r.AddProducer<KafkaMessage>(Topic, producerConfig);
-
                         r.UsingKafka((_, k) =>
                         {
                             k.TopicEndpoint<KafkaMessage>(Topic, consumerConfig, c =>