RabbitMQProvider.cs improvements

Author: Pavlo Korobov

src/providers/WorkflowCore.QueueProviders.RabbitMQ/Interfaces/IRabbitMqQueueNameProvider.cs

@@ -0,0 +1,11 @@
+using System;
+using System.Linq;
+using WorkflowCore.Interface;
+
+namespace WorkflowCore.QueueProviders.RabbitMQ.Interfaces
+{
+    public interface IRabbitMqQueueNameProvider
+    {
+        string GetQueueName(QueueType queue);
+    }
+}

src/providers/WorkflowCore.QueueProviders.RabbitMQ/ServiceCollectionExtensions.cs

@@ -2,18 +2,52 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
-using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using WorkflowCore.Interface;
 using WorkflowCore.Models;
+using WorkflowCore.QueueProviders.RabbitMQ.Interfaces;
 using WorkflowCore.QueueProviders.RabbitMQ.Services;
 
 namespace Microsoft.Extensions.DependencyInjection
 {
+    public delegate IConnection RabbitMqConnectionFactory(IServiceProvider sp, string clientProvidedName);
+    
     public static class ServiceCollectionExtensions
     {
         public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options, IConnectionFactory connectionFactory)
         {
-            options.UseQueueProvider(sp => new RabbitMQProvider(connectionFactory));
+            if (options == null) throw new ArgumentNullException(nameof(options));
+            if (connectionFactory == null) throw new ArgumentNullException(nameof(connectionFactory));
+
+            return options
+                .UseRabbitMQ((sp, name) => connectionFactory.CreateConnection(name));
+        }
+        
+        public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options,
+            IConnectionFactory connectionFactory,
+            IEnumerable<string> hostnames)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+            if (connectionFactory == null) throw new ArgumentNullException(nameof(connectionFactory));
+            if (hostnames == null) throw new ArgumentNullException(nameof(hostnames));
+
+            return options
+                .UseRabbitMQ((sp, name) => connectionFactory.CreateConnection(hostnames.ToList(), name));
+        }
+        
+        public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options, RabbitMqConnectionFactory rabbitMqConnectionFactory)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+            if (rabbitMqConnectionFactory == null) throw new ArgumentNullException(nameof(rabbitMqConnectionFactory));
+
+            options.Services.AddSingleton(rabbitMqConnectionFactory);
+            options.Services.TryAddTransient<IRabbitMqQueueNameProvider, DefaultRabbitMqQueueNameProvider>();
+            options.UseQueueProvider(RabbitMqQueueProviderFactory);
+            
             return options;
         }
+
+        private static IQueueProvider RabbitMqQueueProviderFactory(IServiceProvider sp)
+            => new RabbitMQProvider(sp);
     }
 }

src/providers/WorkflowCore.QueueProviders.RabbitMQ/Services/DefaultRabbitMqQueueNameProvider.cs

@@ -0,0 +1,23 @@
+using WorkflowCore.Interface;
+using WorkflowCore.QueueProviders.RabbitMQ.Interfaces;
+
+namespace WorkflowCore.QueueProviders.RabbitMQ.Services
+{
+    public class DefaultRabbitMqQueueNameProvider : IRabbitMqQueueNameProvider
+    {
+        public string GetQueueName(QueueType queue)
+        {
+            switch (queue)
+            {
+                case QueueType.Workflow:
+                    return "wfc.workflow_queue";
+                case QueueType.Event:
+                    return "wfc.event_queue";
+                case QueueType.Index:
+                    return "wfc.index_queue";
+                default:
+                    return null;
+            }
+        }
+    }
+}

src/providers/WorkflowCore.QueueProviders.RabbitMQ/Services/RabbitMQProvider.cs

@@ -7,23 +7,30 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
 using WorkflowCore.Interface;
 using WorkflowCore.Models;
+using WorkflowCore.QueueProviders.RabbitMQ.Interfaces;
 
 namespace WorkflowCore.QueueProviders.RabbitMQ.Services
 {
 #pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
     public class RabbitMQProvider : IQueueProvider
     {
-        private readonly IConnectionFactory _connectionFactory;
+        private readonly IRabbitMqQueueNameProvider _queueNameProvider;
+        private readonly RabbitMqConnectionFactory _rabbitMqConnectionFactory;
+        private readonly IServiceProvider _serviceProvider;
+        
         private IConnection _connection = null;
         private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings() { TypeNameHandling = TypeNameHandling.All };
 
         public bool IsDequeueBlocking => false;
 
-        public RabbitMQProvider(IConnectionFactory connectionFactory)
+        public RabbitMQProvider(IServiceProvider serviceProvider)
         {
-            _connectionFactory = connectionFactory;
+            _serviceProvider = serviceProvider;
+            _queueNameProvider = _serviceProvider.GetRequiredService<IRabbitMqQueueNameProvider>();
+            _rabbitMqConnectionFactory = _serviceProvider.GetRequiredService<RabbitMqConnectionFactory>();
         }
 
         public async Task QueueWork(string id, QueueType queue)
@@ -33,9 +40,9 @@ public async Task QueueWork(string id, QueueType queue)
 
             using (var channel = _connection.CreateModel())
             {
-                channel.QueueDeclare(queue: GetQueueName(queue), durable: true, exclusive: false, autoDelete: false, arguments: null);
+                channel.QueueDeclare(queue: _queueNameProvider.GetQueueName(queue), durable: true, exclusive: false, autoDelete: false, arguments: null);
                 var body = Encoding.UTF8.GetBytes(id);
-                channel.BasicPublish(exchange: "", routingKey: GetQueueName(queue), basicProperties: null, body: body);
+                channel.BasicPublish(exchange: "", routingKey: _queueNameProvider.GetQueueName(queue), basicProperties: null, body: body);
             }
         }
 
@@ -46,15 +53,15 @@ public async Task<string> DequeueWork(QueueType queue, CancellationToken cancell
 
             using (var channel = _connection.CreateModel())
             {
-                channel.QueueDeclare(queue: GetQueueName(queue),
+                channel.QueueDeclare(queue: _queueNameProvider.GetQueueName(queue),
                                      durable: true,
                                      exclusive: false,
                                      autoDelete: false,
                                      arguments: null);
 
                 channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
 
-                var msg = channel.BasicGet(GetQueueName(queue), false);
+                var msg = channel.BasicGet(_queueNameProvider.GetQueueName(queue), false);
                 if (msg != null)
                 {
                     var data = Encoding.UTF8.GetString(msg.Body);
@@ -76,7 +83,7 @@ public void Dispose()
 
         public async Task Start()
         {
-            _connection = _connectionFactory.CreateConnection("Workflow-Core");
+            _connection = _rabbitMqConnectionFactory(_serviceProvider, "Workflow-Core");
         }
 
         public async Task Stop()
@@ -88,20 +95,6 @@ public async Task Stop()
             }
         }
 
-        private string GetQueueName(QueueType queue)
-        {
-            switch (queue)
-            {
-                case QueueType.Workflow:
-                    return "wfc.workflow_queue";
-                case QueueType.Event:
-                    return "wfc.event_queue";
-                case QueueType.Index:
-                    return "wfc.index_queue";
-            }
-            return null;
-        }
-                
     }
 #pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
 }