Unique name queues

Src/Coravel/QueueServiceRegistration.cs

@@ -1,9 +1,13 @@
 using System;
+using System.Linq;
 using Coravel.Events.Interfaces;
 using Coravel.Queuing;
 using Coravel.Queuing.HostedService;
 using Coravel.Queuing.Interfaces;
+using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
 
 namespace Coravel
 {
@@ -29,6 +33,37 @@ public static IServiceCollection AddQueue(this IServiceCollection services)
             return services;
         }
 
+        public static IServiceCollection AddQueue(this IServiceCollection services, string queueName)
+        {
+            services.AddSingleton<IQueue>(p =>
+                new Queue(
+                    queueName,
+                    p.GetRequiredService<IServiceScopeFactory>(),
+                    p.GetService<IDispatcher>()
+                )
+            );
+
+            //It is not possible to use services.AddHostedService here.
+            //AddHostedService method only registers the service if it is not one already registered.
+            //Please have a look at:
+            //https://github.com/dotnet/runtime/blob/57bfe474518ab5b7cfe6bf7424a79ce3af9d6657/src/libraries/Microsoft.Extensions.Hosting.Abstractions/src/ServiceCollectionHostedServiceExtensions.cs
+            services.AddSingleton<IHostedService, QueuingHost>(p =>
+            {
+                var queue = p.GetServices<IQueue>().First(x => x.QueueName == queueName);
+                var configuration = p.GetRequiredService<IConfiguration>();
+                var logger = p.GetRequiredService<ILogger<QueuingHost>>();
+                return new QueuingHost(queue, configuration, logger);
+            });
+            return services;
+        }
+
+        public static IServiceCollection AddQueues(this IServiceCollection services, Action<IServiceCollection> registerQueues)
+        {
+            registerQueues(services);
+            services.AddSingleton<IQueues, QueuesCollection>();
+            return services;
+        }
+
         public static IQueueConfiguration ConfigureQueue(this IServiceProvider provider)
         {
             var queue = provider.GetRequiredService<IQueue>();

Src/Coravel/Queuing/Exceptions/QueueNotFoundException.cs

@@ -0,0 +1,12 @@
+using System;
+
+namespace Coravel.Queuing.Exceptions
+{
+    public class QueueNotFoundException : Exception
+    {
+        public QueueNotFoundException(string name): base($"Queue with name '{name}' doesn't exist!")
+        {
+
+        }
+    }
+}

Src/Coravel/Queuing/HostedService/QueuingHost.cs

@@ -29,7 +29,10 @@ public QueuingHost(IQueue queue, IConfiguration configuration, ILogger<QueuingHo
         public Task StartAsync(CancellationToken cancellationToken)
         {
             int consummationDelay = GetConsummationDelay();
-
+            if (!string.IsNullOrEmpty(_queue.QueueName))
+            {
+                _logger.LogDebug($"Coravel is starting queue service for '{_queue.QueueName}'");
+            }
             this._timer = new Timer((state) => this._signal.Release(), null, TimeSpan.Zero, TimeSpan.FromSeconds(consummationDelay));
             Task.Run(ConsumeQueueAsync);
             return Task.CompletedTask;
@@ -74,7 +77,14 @@ public async Task StopAsync(CancellationToken cancellationToken)
         public void Dispose()
         {
             this._timer?.Dispose();
-            this._logger.LogInformation("Coravel's Queuing service is now stopped.");
+            if (string.IsNullOrEmpty(_queue.QueueName))
+            {
+                this._logger.LogInformation("Coravel's Queuing service is now stopped.");
+            }
+            else
+            {
+                this._logger.LogInformation($"Coravel's Queuing service for '{_queue.QueueName}' is now stopped.");
+            }
         }
     }
 }

Src/Coravel/Queuing/Interfaces/IQueue.cs

@@ -11,6 +11,10 @@ namespace Coravel.Queuing.Interfaces
     /// </summary>
     public interface IQueue
     {
+        /// <summary>
+        /// Contains a unique name for the queue!
+        /// </summary>
+        string QueueName { get; }
         /// <summary>
         /// Queue a new synchronous task.
         /// </summary>

Src/Coravel/Queuing/Interfaces/IQueues.cs

@@ -0,0 +1,9 @@
+using System.Collections;
+
+namespace Coravel.Queuing.Interfaces
+{
+    public interface IQueues
+    {
+        IQueue Get(string queueName);
+    }
+}

Src/Coravel/Queuing/Queue.cs

@@ -16,6 +16,7 @@ namespace Coravel.Queuing
 {
     public class Queue : IQueue, IQueueConfiguration
     {
+        public string QueueName { get; private set; }
         private ConcurrentQueue<ActionOrAsyncFunc> _tasks = new ConcurrentQueue<ActionOrAsyncFunc>();
         private ConcurrentDictionary<Guid, CancellationTokenSource> _tokens = new ConcurrentDictionary<Guid, CancellationTokenSource>();
         private Action<Exception> _errorHandler;
@@ -32,6 +33,12 @@ public Queue(IServiceScopeFactory scopeFactory, IDispatcher dispatcher)
             this._dispatcher = dispatcher;
         }
 
+        public Queue(string queueName, IServiceScopeFactory scopeFactory, IDispatcher dispatcher) {
+            this.QueueName = queueName;
+            this._scopeFactory = scopeFactory;
+            this._dispatcher = dispatcher;
+        }
+
         public Guid QueueTask(Action task)
         {
             var job = new ActionOrAsyncFunc(task);
@@ -115,7 +122,7 @@ await Task.WhenAll(
             }
         }
 
-        public async Task ConsumeQueueOnShutdown() 
+        public async Task ConsumeQueueOnShutdown()
         {
             this.CancelAllTokens();
             await this.ConsumeQueueAsync();
@@ -153,9 +160,9 @@ private ActionOrAsyncFunc EnqueueInvocable<T>(Action<IInvocable> beforeInvoked =
                     using (var scope = this._scopeFactory.CreateScope())
                     {
                         if (scope.ServiceProvider.GetService(invocableType) is IInvocable invocable)
-                        {                            
+                        {
                             if(beforeInvoked != null)
-                            {                            
+                            {
                                 beforeInvoked(invocable);
                             }
 
@@ -180,7 +187,7 @@ private void CleanTokens(IEnumerable<Guid> guidsForTokensToClean)
                 {
                     token.Dispose();
                 }
-            }                   
+            }
         }
 
         private List<ActionOrAsyncFunc> DequeueAllTasks()

Src/Coravel/Queuing/QueuesCollection.cs

@@ -0,0 +1,26 @@
+using System.Collections;
+using System.Collections.Generic;
+using Coravel.Queuing.Interfaces;
+using System.Linq;
+using Coravel.Queuing.Exceptions;
+
+namespace Coravel.Queuing
+{
+    public class QueuesCollection : IQueues
+    {
+        private readonly IEnumerable<IQueue> _queues;
+
+        public QueuesCollection(IEnumerable<IQueue> queues)
+        {
+            _queues = queues;
+        }
+        public IQueue Get(string queueName)
+        {
+            if (_queues.FirstOrDefault(x => x.QueueName == queueName) is IQueue queue)
+            {
+                return queue;
+            }
+            throw new QueueNotFoundException(queueName);
+        }
+    }
+}