Implement a CombinedProducerRegistryFactory for using multiple messag…

…e transports [v9] (#3273)

* Start of message producer factory implementations

* Finish message producer factory implementations

* Add in memory message producer factory & test

* Fix backporting issues

* Fix unimplemented dispose method

src/Paramore.Brighter.MessagingGateway.AWSSQS/SnsMessageProducerFactory.cs

@@ -0,0 +1,62 @@
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2024 Dominic Hickie <[email protected]>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+#endregion
+
+using System.Collections.Generic;
+
+namespace Paramore.Brighter.MessagingGateway.AWSSQS
+{
+    public class SnsMessageProducerFactory : IAmAMessageProducerFactory
+    {
+        private readonly AWSMessagingGatewayConnection _connection;
+        private readonly IEnumerable<SnsPublication> _snsPublications;
+
+        /// <summary>
+        /// Creates a collection of SNS message producers from the SNS publication information
+        /// </summary>
+        /// <param name="connection">The Connection to use to connect to AWS</param>
+        /// <param name="snsPublications">The publications describing the SNS topics that we want to use</param>
+        public SnsMessageProducerFactory(
+            AWSMessagingGatewayConnection connection,
+            IEnumerable<SnsPublication> snsPublications)
+        {
+            _connection = connection;
+            _snsPublications = snsPublications;
+        }
+
+        /// <inheritdoc />
+        public Dictionary<string,IAmAMessageProducer> Create()
+        {
+            var producers = new Dictionary<string, IAmAMessageProducer>();
+            foreach (var p in _snsPublications)
+            {
+                var producer = new SqsMessageProducer(_connection, p);
+                if (producer.ConfirmTopicExists())
+                    producers[p.Topic] = producer;
+                else
+                    throw new ConfigurationException($"Missing SNS topic: {p.Topic}");
+            }
+
+            return producers;
+        }
+    }
+}

src/Paramore.Brighter.MessagingGateway.AWSSQS/SnsProducerRegistryFactory.cs

@@ -22,8 +22,6 @@ THE SOFTWARE. */
 #endregion
 
 using System.Collections.Generic;
-using Amazon;
-using Amazon.Runtime;
 
 namespace Paramore.Brighter.MessagingGateway.AWSSQS
 {
@@ -51,18 +49,9 @@ public SnsProducerRegistryFactory(
         /// <returns></returns>
         public IAmAProducerRegistry Create()
         {
-            var producers = new Dictionary<string, IAmAMessageProducer>();
-            foreach (var p in _snsPublications)
-            {
-                var producer = new SqsMessageProducer(_connection, p);
-                if (producer.ConfirmTopicExists())
-                    producers[p.Topic] = producer;
-                else
-                    throw new ConfigurationException($"Missing SNS topic: {p.Topic}");
+            var producerFactory = new SnsMessageProducerFactory(_connection, _snsPublications);
 
-            }
-
-            return new ProducerRegistry(producers);
+            return new ProducerRegistry(producerFactory.Create());
         }
     }
 }

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusMessageProducerFactory.cs

@@ -1,43 +1,67 @@
-using Paramore.Brighter.MessagingGateway.AzureServiceBus.AzureServiceBusWrappers;
+using System.Collections.Generic;
+using Paramore.Brighter.MessagingGateway.AzureServiceBus.AzureServiceBusWrappers;
 using Paramore.Brighter.MessagingGateway.AzureServiceBus.ClientProvider;
 
 namespace Paramore.Brighter.MessagingGateway.AzureServiceBus
 {
     /// <summary>
-    /// Factory class for creating instances of <see cref="AzureServiceBusMessageProducer"/>
+    /// Factory class for creating dictionary of instances of <see cref="AzureServiceBusMessageProducer"/>
+    /// indexed by topic name
     /// </summary>
-    internal static class AzureServiceBusMessageProducerFactory
+    public class AzureServiceBusMessageProducerFactory : IAmAMessageProducerFactory
     {
+        private readonly IServiceBusClientProvider _clientProvider;
+        private readonly IEnumerable<AzureServiceBusPublication> _publications;
+        private readonly int _bulkSendBatchSize;
+
         /// <summary>
-        /// Factory to create an Azure Service Bus Producer
+        /// Factory to create a dictionary of Azure Service Bus Producers indexed by topic name
         /// </summary>
-        /// <param name="configuration">The configuration to connect to <see cref="AzureServiceBusConfiguration"/></param>
-        /// <param name="asbPublication">Describes the parameters for the producer</param>
-        /// <returns>A Message Producer</returns>
-        public static AzureServiceBusMessageProducer Get(
+        /// <param name="configuration">The configuration of the connection to ASB</param>
+        /// <param name="publications">A set of publications - topics on the server - to configure</param>
+        public AzureServiceBusMessageProducerFactory(
             AzureServiceBusConfiguration configuration,
-            AzureServiceBusPublication asbPublication)
+            IEnumerable<AzureServiceBusPublication> publications)
         {
-            var clientProvider = new ServiceBusConnectionStringClientProvider(configuration.ConnectionString);
-            return Get(clientProvider, asbPublication, configuration.BulkSendBatchSize);
+            _clientProvider = new ServiceBusConnectionStringClientProvider(configuration.ConnectionString);
+            _publications = publications;
+            _bulkSendBatchSize = configuration.BulkSendBatchSize;
         }
 
         /// <summary>
-        /// Factory to create an Azure Service Bus Producer
+        /// Factory to create a dictionary of Azure Service Bus Producers indexed by topic name
         /// </summary>
         /// <param name="clientProvider">The connection to ASB</param>
-        /// <param name="asbPublication">Describes the parameters for the producer</param>
-        /// <param name="bulkSendBatchSize">When sending more than one message using the MessageProducer, the max amount to send in a single transmission.</param>
-        /// <returns></returns>
-        public static AzureServiceBusMessageProducer Get(
+        /// <param name="publications">A set of publications - topics on the server - to configure</param>
+        /// <param name="bulkSendBatchSize">The maximum size to chunk messages when dispatching to ASB</param>
+        public AzureServiceBusMessageProducerFactory(
             IServiceBusClientProvider clientProvider,
-            AzureServiceBusPublication asbPublication,
-            int bulkSendBatchSize = 10)
+            IEnumerable<AzureServiceBusPublication> publications,
+            int bulkSendBatchSize)
         {
-            var nameSpaceManagerWrapper = new AdministrationClientWrapper(clientProvider);
-            var topicClientProvider = new ServiceBusSenderProvider(clientProvider);
-
-            return new AzureServiceBusMessageProducer(nameSpaceManagerWrapper, topicClientProvider, asbPublication, bulkSendBatchSize);
+            _clientProvider = clientProvider;
+            _publications = publications;
+            _bulkSendBatchSize = bulkSendBatchSize;
         }
+
+        /// <inheritdoc />
+        public Dictionary<string, IAmAMessageProducer> Create()
+        {
+            var nameSpaceManagerWrapper = new AdministrationClientWrapper(_clientProvider);
+            var topicClientProvider = new ServiceBusSenderProvider(_clientProvider);
+
+            var producers = new Dictionary<string, IAmAMessageProducer>();
+            foreach (var publication in _publications)
+            {
+                var producer = new AzureServiceBusMessageProducer(
+                    nameSpaceManagerWrapper, 
+                    topicClientProvider, 
+                    publication, 
+                    _bulkSendBatchSize);
+                producers.Add(publication.Topic, producer);
+            }
+
+            return producers;
+        }        
     }
 }

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusProducerRegistryFactory.cs

@@ -45,13 +45,9 @@ public AzureServiceBusProducerRegistryFactory(
         /// <returns>A has of middleware clients by topic, for sending messages to the middleware</returns>
         public IAmAProducerRegistry Create()
         {
-            var producers = new Dictionary<string, IAmAMessageProducer>();
-            foreach (var publication in _asbPublications)
-            {
-                producers[publication.Topic] = AzureServiceBusMessageProducerFactory.Get(_clientProvider, publication, _bulkSendBatchSize);
-            }
+            var producerFactory = new AzureServiceBusMessageProducerFactory(_clientProvider, _asbPublications, _bulkSendBatchSize);
 
-            return new ProducerRegistry(producers);
+            return new ProducerRegistry(producerFactory.Create());
         }
     }
 }

src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessageProducerFactory.cs

@@ -0,0 +1,85 @@
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2024 Dominic Hickie <[email protected]>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE. */
+#endregion
+
+using System;
+using System.Collections.Generic;
+using Confluent.Kafka;
+
+namespace Paramore.Brighter.MessagingGateway.Kafka
+{
+    /// <summary>
+    /// Creates a dictionary of <see cref="KafkaMessageProducer"/> instances indexed by topic from a collection of <see cref="Publication"/> instances
+    /// Note that we only return the interface and <see cref="KafkaMessageProducer"/> is internal as the underlying type is not needed
+    /// </summary>
+    public class KafkaMessageProducerFactory : IAmAMessageProducerFactory
+    {
+        private readonly KafkaMessagingGatewayConfiguration _globalConfiguration;
+        private readonly IEnumerable<KafkaPublication> _publications;
+        private Action<ProducerConfig> _configHook;
+
+        /// <summary>
+        /// This constructs a <see cref="KafkaMessageProducerFactory"/> which can be used to create a dictionary of <see cref="KafkaMessageProducer"/>
+        /// instances indexed by topic name.
+        /// It takes a dependency on a <see cref="KafkaMessagingGatewayConfiguration"/> to connect to the broker, and a collection of 
+        /// <see cref="KafkaPublication"/> instances that determine how we publish to Kafka and the parameters of any topics if required.
+        /// </summary>
+        /// <param name="globalConfiguration">Configures how we connect to the broker</param>
+        /// <param name="publications">The list of topics that we want to publish to</param>
+        public KafkaMessageProducerFactory(
+            KafkaMessagingGatewayConfiguration globalConfiguration, 
+            IEnumerable<KafkaPublication> publications)
+        {
+            _globalConfiguration = globalConfiguration;
+            _publications = publications;
+            _configHook = null;
+        }
+
+        /// <inheritdoc />
+        public Dictionary<string,IAmAMessageProducer> Create()
+        {
+            var publicationsByTopic = new Dictionary<string, IAmAMessageProducer>();
+            foreach (var publication in _publications)
+            {
+
+                var producer = new KafkaMessageProducer(_globalConfiguration, publication);
+                if (_configHook != null)
+                    producer.ConfigHook(_configHook);
+                producer.Init();
+                publicationsByTopic[publication.Topic] = producer;
+            }
+
+            return publicationsByTopic;
+        }
+
+        /// <summary>
+        /// Set a configuration hook to set properties not exposed by KafkaMessagingGatewayConfiguration or KafkaPublication
+        /// Intended as 'get out of gaol free' this couples us to the Confluent .NET Kafka client. Bear in mind that a future release
+        /// might drop the Confluent client, and this hook
+        /// </summary>
+        /// <param name="hook"></param>
+        public void SetConfigHook(Action<ProducerConfig> hook)
+        {
+            _configHook = hook;
+        }
+    }
+}

src/Paramore.Brighter.MessagingGateway.Kafka/KafkaProducerRegistryFactory.cs

@@ -28,7 +28,7 @@ THE SOFTWARE. */
 namespace Paramore.Brighter.MessagingGateway.Kafka
 {
     /// <summary>
-    /// Creates a <see cref="KafkaMessageProducer"/> from a <see cref="Publication"/>
+    /// Creates a registry of <see cref="KafkaMessageProducer"/> instances from <see cref="Publication"/> instances
     /// Note that we only return the interface and <see cref="KafkaMessageProducer"/> is internal as the underlying type is not needed
     /// </summary>
     public class KafkaProducerRegistryFactory : IAmAProducerRegistryFactory
@@ -38,9 +38,9 @@ public class KafkaProducerRegistryFactory : IAmAProducerRegistryFactory
         private Action<ProducerConfig> _configHook;
 
         /// <summary>
-        /// This constructs a <see cref="KafkaProducerRegistryFactory"/> which can be used to create a <see cref="KafkaMessageProducer"/>.
-        /// It takes a dependency on a <see cref="KafkaMessagingGatewayConfiguration"/> to connect to the broker, and a <see cref="KafkaPublication"/>
-        /// that determines how we publish to Kafka and the parameters of any topic if required.
+        /// This constructs a <see cref="KafkaProducerRegistryFactory"/> which can be used to create a <see cref="ProducerRegistry" /> of <see cref="KafkaMessageProducer" /> instances.
+        /// It takes a dependency on a <see cref="KafkaMessagingGatewayConfiguration"/> to connect to the broker, and a collection of <see cref="KafkaPublication"/> instances
+        /// that determine how we publish to Kafka and the parameters of any topics if required.
         /// </summary>
         /// <param name="globalConfiguration">Configures how we connect to the broker</param>
         /// <param name="publication">How do we publish, both producer parameters and topic configuration</param>
@@ -54,24 +54,16 @@ public KafkaProducerRegistryFactory(
         }
 
         /// <summary>
-        /// Create a message producer from tne <see cref="KafkaMessagingGatewayConfiguration"/> and <see cref="KafkaPublication"/> supplied
+        /// Create a producer registry from the <see cref="KafkaMessagingGatewayConfiguration"/> and <see cref="KafkaPublication"/> instances supplied
         /// to the constructor
         /// </summary>
         /// <returns>An <see cref="IAmAProducerRegistry"/> that represents a collection of Kafka Message Producers</returns>
         public IAmAProducerRegistry Create()
         {
-            var publicationsByTopic = new Dictionary<string, IAmAMessageProducer>();
-            foreach (var publication in _publications)
-            {
+            var producerFactory = new KafkaMessageProducerFactory(_globalConfiguration, _publications);
+            producerFactory.SetConfigHook(_configHook);
 
-                var producer = new KafkaMessageProducer(_globalConfiguration, publication);
-                if (_configHook != null)
-                    producer.ConfigHook(_configHook);
-                producer.Init();
-                publicationsByTopic[publication.Topic] = producer;
-            }
-
-            return new ProducerRegistry(publicationsByTopic);
+            return new ProducerRegistry(producerFactory.Create());
         }
 
         /// <summary>