Refactored channel pool and configuration.

framework/src/Volo.Abp.BackgroundJobs.Abstractions/Volo/Abp/BackgroundJobs/BackgroundJobOptions.cs

@@ -21,13 +21,18 @@ namespace Volo.Abp.BackgroundJobs
             _jobConfigurationsByName = new Dictionary<string, BackgroundJobConfiguration>();
         }
 
+        public BackgroundJobConfiguration GetJob<TArgs>()
+        {
+            return GetJob(typeof(TArgs));
+        }
+
         public BackgroundJobConfiguration GetJob(Type argsType)
         {
             var jobConfiguration = _jobConfigurationsByArgsType.GetOrDefault(argsType);
 
             if (jobConfiguration == null)
             {
-                throw new AbpException("Undefined background job type for the job args type: " + argsType.AssemblyQualifiedName);
+                throw new AbpException("Undefined background job for the job args type: " + argsType.AssemblyQualifiedName);
             }
 
             return jobConfiguration;
@@ -39,7 +44,7 @@ namespace Volo.Abp.BackgroundJobs
 
             if (jobConfiguration == null)
             {
-                throw new AbpException("Undefined background job type for the job name: " + name);
+                throw new AbpException("Undefined background job for the job name: " + name);
             }
 
             return jobConfiguration;

framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs

@@ -16,9 +16,10 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
 
     public class JobQueue<TArgs> : IJobQueue<TArgs>
     {
-        protected Type JobType { get; }
-        protected string JobName { get; }
-        protected string QueueName { get; }
+        private const string ChannelPrefix = "JobQueue.";
+
+        protected BackgroundJobConfiguration JobConfiguration { get; }
+        protected JobQueueConfiguration QueueConfiguration { get; }
 
         protected IChannelAccessor ChannelAccessor { get; private set; }
         protected EventingBasicConsumer Consumer { get; private set; }
@@ -30,6 +31,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
         protected BackgroundJobOptions BackgroundJobOptions { get; }
         protected IRabbitMqSerializer Serializer { get; }
         protected IBackgroundJobExecuter JobExecuter { get; }
+        protected RabbitMqBackgroundJobOptions RabbitMqBackgroundJobOptions { get; }
 
         protected AsyncLock SyncObj = new AsyncLock();
         protected bool IsDiposed { get; private set; }
@@ -37,19 +39,25 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
         public JobQueue(
             IChannelPool channelPool,
             IRabbitMqSerializer serializer,
-            IOptions<AbpRabbitMqOptions> options,
+            IOptions<AbpRabbitMqOptions> rabbitMqOptions,
             IBackgroundJobExecuter jobExecuter,
-            IOptions<BackgroundJobOptions> backgroundJobOptions)
+            IOptions<BackgroundJobOptions> backgroundJobOptions,
+            IOptions<RabbitMqBackgroundJobOptions> rabbitMqBackgroundJobOptions)
         {
             Serializer = serializer;
             JobExecuter = jobExecuter;
             BackgroundJobOptions = backgroundJobOptions.Value;
             ChannelPool = channelPool;
-            RabbitMqOptions = options.Value;
+            RabbitMqOptions = rabbitMqOptions.Value;
+            RabbitMqBackgroundJobOptions = rabbitMqBackgroundJobOptions.Value;
+
+            JobConfiguration = BackgroundJobOptions.GetJob(typeof(TArgs));
 
-            JobName = BackgroundJobNameAttribute.GetName<TArgs>();
-            JobType = BackgroundJobOptions.GetJob(typeof(TArgs)).JobType;
-            QueueName = "BackgroundJobs." + JobName; //TODO: Make prefix optional
+            QueueConfiguration = RabbitMqBackgroundJobOptions.JobQueues.GetOrDefault(typeof(TArgs)) ??
+                            new JobQueueConfiguration(
+                                typeof(TArgs),
+                                RabbitMqBackgroundJobOptions.DefaultQueueNamePrefix + JobConfiguration.JobName
+                            );
 
             Logger = NullLogger<JobQueue<TArgs>>.Instance;
         }
@@ -111,13 +119,13 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
                 return Task.CompletedTask;
             }
 
-            ChannelAccessor = ChannelPool.Acquire(QueueName + ".JobQueue");
-
-            var queueOptions = RabbitMqOptions.Queues.GetOrDefault(QueueName)
-                               ?? new QueueOptions(QueueName);
+            ChannelAccessor = ChannelPool.Acquire(
+                ChannelPrefix + QueueConfiguration.QueueName,
+                QueueConfiguration.ConnectionName
+            );
 
-            var result = queueOptions.Declare(ChannelAccessor.Channel);
-            Logger.LogDebug($"RabbitMQ Queue '{QueueName}' has {result.MessageCount} messages and {result.ConsumerCount} consumers.");
+            var result = QueueConfiguration.Declare(ChannelAccessor.Channel);
+            Logger.LogDebug($"RabbitMQ Queue '{QueueConfiguration.QueueName}' has {result.MessageCount} messages and {result.ConsumerCount} consumers.");
 
             if (BackgroundJobOptions.IsJobExecutionEnabled)
             {
@@ -126,7 +134,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
 
                 //TODO: What BasicConsume returns?
                 ChannelAccessor.Channel.BasicConsume(
-                    queue: QueueName,
+                    queue: QueueConfiguration.QueueName,
                     autoAck: false,
                     consumer: Consumer
                 );
@@ -144,7 +152,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
 
             ChannelAccessor.Channel.BasicPublish(
                 exchange: "",
-                routingKey: QueueName,
+                routingKey: QueueConfiguration.QueueName,
                 basicProperties: CreateBasicPropertiesToPublish(),
                 body: Serializer.Serialize(args)
             );
@@ -162,7 +170,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
         protected virtual void MessageReceived(object sender, BasicDeliverEventArgs ea)
         {
             var context = new JobExecutionContext(
-                JobType,
+                JobConfiguration.JobType,
                 Serializer.Deserialize(ea.Body, typeof(TArgs))
             );
 

framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs

@@ -0,0 +1,29 @@
+using System;
+using Volo.Abp.RabbitMQ;
+
+namespace Volo.Abp.BackgroundJobs.RabbitMQ
+{
+    public class JobQueueConfiguration : QueueConfiguration
+    {
+        public Type JobArgsType { get; }
+
+        public string ConnectionName { get; set; }
+
+        public JobQueueConfiguration(
+            Type jobArgsType, 
+            string queueName, 
+            string connectionName = null,
+            bool durable = true,
+            bool exclusive = false,
+            bool autoDelete = false)
+            : base(
+                queueName, 
+                durable, 
+                exclusive, 
+                autoDelete)
+        {
+            JobArgsType = jobArgsType;
+            ConnectionName = connectionName;
+        }
+    }
+}

framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/RabbitMqBackgroundJobOptions.cs

@@ -0,0 +1,24 @@
+using System;
+using System.Collections.Generic;
+
+namespace Volo.Abp.BackgroundJobs.RabbitMQ
+{
+    public class RabbitMqBackgroundJobOptions
+    {
+        /// <summary>
+        /// Key: Job Args Type
+        /// </summary>
+        public Dictionary<Type, JobQueueConfiguration> JobQueues { get; }
+
+        /// <summary>
+        /// Default value: "AbpBackgroundJobs.".
+        /// </summary>
+        public string DefaultQueueNamePrefix { get; set; }
+
+        public RabbitMqBackgroundJobOptions()
+        {
+            JobQueues = new Dictionary<Type, JobQueueConfiguration>();
+            DefaultQueueNamePrefix = "AbpBackgroundJobs.";
+        }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqOptions.cs

@@ -4,12 +4,9 @@
     {
         public RabbitMqConnections ConnectionFactories { get; }
 
-        public QueueOptionsDictionary Queues { get; }
-
         public AbpRabbitMqOptions()
         {
             ConnectionFactories = new RabbitMqConnections();
-            Queues = new QueueOptionsDictionary();
         }
     }
 }

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs

@@ -29,7 +29,7 @@ namespace Volo.Abp.RabbitMQ
             Logger = NullLogger<ChannelPool>.Instance;
         }
 
-        public virtual IChannelAccessor Acquire(string channelName = null)
+        public virtual IChannelAccessor Acquire(string channelName = null, string connectionName = null)
         {
             CheckDisposed();
 
@@ -37,21 +37,23 @@ namespace Volo.Abp.RabbitMQ
 
             var poolItem = Channels.GetOrAdd(
                 channelName,
-                _ => new ChannelPoolItem(CreateChannel(channelName))
+                _ => new ChannelPoolItem(CreateChannel(channelName, connectionName))
             );
 
             poolItem.Acquire();
 
             return new ChannelAccessor(
                 poolItem.Channel,
+                channelName,
                 () => poolItem.Release()
             );
         }
 
-        protected virtual IModel CreateChannel(string channelName)
+        protected virtual IModel CreateChannel(string channelName, string connectionName)
         {
-            //TODO: How to determine the right connection name?
-            return ConnectionPool.Get().CreateModel();
+            return ConnectionPool
+                .Get(connectionName)
+                .CreateModel();
         }
 
         protected void CheckDisposed()
@@ -174,11 +176,15 @@ namespace Volo.Abp.RabbitMQ
         protected class ChannelAccessor : IChannelAccessor
         {
             public IModel Channel { get; }
+
+            public string Name { get; }
+
             private readonly Action _disposeAction;
 
-            public ChannelAccessor(IModel channel, Action disposeAction)
+            public ChannelAccessor(IModel channel, string name, Action disposeAction)
             {
                 _disposeAction = disposeAction;
+                Name = name;
                 Channel = channel;
             }
 

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IChannelAccessor.cs

@@ -11,5 +11,10 @@ namespace Volo.Abp.RabbitMQ
         /// Instead, dispose the <see cref="IChannelAccessor"/> after usage.
         /// </summary>
         IModel Channel { get; }
+
+        /// <summary>
+        /// Name of the channel.
+        /// </summary>
+        string Name { get; }
     }
 }

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IChannelPool.cs

@@ -4,6 +4,6 @@ namespace Volo.Abp.RabbitMQ
 {
     public interface IChannelPool : IDisposable
     {
-        IChannelAccessor Acquire(string channelName = null);
+        IChannelAccessor Acquire(string channelName = null, string connectionName = null);
     }
 }

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueOptions.cs → framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueConfiguration.cs

@@ -4,10 +4,10 @@ using RabbitMQ.Client;
 
 namespace Volo.Abp.RabbitMQ
 {
-    public class QueueOptions
+    public class QueueConfiguration
     {
         [NotNull]
-        public string Name { get; }
+        public string QueueName { get; }
 
         public bool Durable { get; set; }
 
@@ -17,23 +17,23 @@ namespace Volo.Abp.RabbitMQ
 
         public IDictionary<string, object> Arguments { get; }
 
-        public QueueOptions(
-            [NotNull] string name, 
+        public QueueConfiguration(
+            [NotNull] string queueName, 
             bool durable = true, 
             bool exclusive = false, 
             bool autoDelete = false)
         {
-            Name = name;
+            QueueName = queueName;
             Durable = durable;
             Exclusive = exclusive;
             AutoDelete = autoDelete;
             Arguments = new Dictionary<string, object>();
         }
 
-        public QueueDeclareOk Declare(IModel channel)
+        public virtual QueueDeclareOk Declare(IModel channel)
         {
             return channel.QueueDeclare(
-                queue: Name,
+                queue: QueueName,
                 durable: Durable,
                 exclusive: Exclusive,
                 autoDelete: AutoDelete,

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueOptionsDictionary.cs

@@ -1,9 +0,0 @@
-using System.Collections.Generic;
-
-namespace Volo.Abp.RabbitMQ
-{
-    public class QueueOptionsDictionary : Dictionary<string, QueueOptions>
-    {
-
-    }
-}

modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp.Shared/Jobs/WriteToConsoleGreenJob.cs

@@ -12,14 +12,17 @@ namespace Volo.Abp.BackgroundJobs.DemoApp.Shared.Jobs
                 //throw new ApplicationException("A sample exception from the WriteToConsoleGreenJob!");
             }
 
-            var oldColor = Console.ForegroundColor;
-            Console.ForegroundColor = ConsoleColor.Green;
+            lock (Console.Out)
+            {
+                var oldColor = Console.ForegroundColor;
+                Console.ForegroundColor = ConsoleColor.Green;
 
-            Console.WriteLine();
-            Console.WriteLine($"############### WriteToConsoleGreenJob: {args.Value} - {args.Time:HH:mm:ss} ###############");
-            Console.WriteLine();
+                Console.WriteLine();
+                Console.WriteLine($"############### WriteToConsoleGreenJob: {args.Value} - {args.Time:HH:mm:ss} ###############");
+                Console.WriteLine();
 
-            Console.ForegroundColor = oldColor;
+                Console.ForegroundColor = oldColor;
+            }
         }
     }
 }

modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp.Shared/Jobs/WriteToConsoleYellowJob.cs

@@ -12,14 +12,17 @@ namespace Volo.Abp.BackgroundJobs.DemoApp.Shared.Jobs
                 //throw new ApplicationException("A sample exception from the WriteToConsoleYellowJob!");
             }
 
-            var oldColor = Console.ForegroundColor;
-            Console.ForegroundColor = ConsoleColor.Yellow;
+            lock (Console.Out)
+            {
+                var oldColor = Console.ForegroundColor;
+                Console.ForegroundColor = ConsoleColor.Yellow;
 
-            Console.WriteLine();
-            Console.WriteLine($"############### WriteToConsoleYellowJob: {args.Value} - {args.Time:HH:mm:ss}  ###############");
-            Console.WriteLine();
+                Console.WriteLine();
+                Console.WriteLine($"############### WriteToConsoleYellowJob: {args.Value} - {args.Time:HH:mm:ss}  ###############");
+                Console.WriteLine();
 
-            Console.ForegroundColor = oldColor;
+                Console.ForegroundColor = oldColor;
+            }
         }
     }
 }