Introcude JobQueue

framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo.Abp.BackgroundJobs.RabbitMQ.csproj

@@ -16,6 +16,7 @@
   <ItemGroup>
     <ProjectReference Include="..\Volo.Abp.BackgroundJobs.Abstractions\Volo.Abp.BackgroundJobs.Abstractions.csproj" />
     <ProjectReference Include="..\Volo.Abp.RabbitMQ\Volo.Abp.RabbitMQ.csproj" />
+    <ProjectReference Include="..\Volo.Abp.Threading\Volo.Abp.Threading.csproj" />
   </ItemGroup>
 
 </Project>

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

@@ -13,7 +13,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
     {
         public override void ConfigureServices(ServiceConfigurationContext context)
         {
-            context.Services.AddSingleton(typeof(JobListener<>)); //TODO: Introduce and use interface!
+            context.Services.AddSingleton(typeof(IJobQueue<>), typeof(JobQueue<>));
 
             context.Services.AddAssemblyOf<AbpBackgroundJobsRabbitMqModule>();
         }

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

@@ -1,8 +0,0 @@
-namespace Volo.Abp.BackgroundJobs.RabbitMQ
-{
-    public interface IJobListener
-    {
-        void Start();
-        void Stop();
-    }
-}

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

@@ -0,0 +1,12 @@
+using System;
+using System.Threading.Tasks;
+
+namespace Volo.Abp.BackgroundJobs.RabbitMQ
+{
+    public interface IJobQueue<TArgs> : IDisposable
+    {
+        Task<string> Enqueue(TArgs args);
+
+        Task StartAsync();
+    }
+}

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

@@ -1,62 +0,0 @@
-using System;
-using RabbitMQ.Client;
-using RabbitMQ.Client.Events;
-using Volo.Abp.RabbitMQ;
-
-namespace Volo.Abp.BackgroundJobs.RabbitMQ
-{
-    public class JobListener<TJob> : IJobListener
-    {
-        protected string JobName { get; }
-        protected IBackgroundJobExecuter JobExecuter { get; }
-        protected IChannelPool ChannelPool { get; }
-        protected IRabbitMqSerializer Serializer { get; }
-        protected Type ArgsType { get; }
-
-        protected IChannelAccessor ChannelAccessor { get; private set; }
-        protected EventingBasicConsumer Consumer { get; private set; }
-
-        public JobListener(
-            IChannelPool channelPool, 
-            IBackgroundJobExecuter jobExecuter, 
-            IRabbitMqSerializer serializer)
-        {
-            ChannelPool = channelPool;
-            JobExecuter = jobExecuter;
-            Serializer = serializer;
-            ArgsType = BackgroundJobArgsHelper.GetJobArgsType(typeof(TJob));
-            JobName = BackgroundJobNameAttribute.GetName(ArgsType);
-        }
-
-        public void Start()
-        {
-            var queueName = "BackgroundJobs." + JobName; //TODO: Make prefix optional
-
-            ChannelAccessor = ChannelPool.Acquire(queueName);
-
-            //TODO: How to ensure that queue is created!
-
-            Consumer = new EventingBasicConsumer(ChannelAccessor.Channel);
-            Consumer.Received += MessageReceived;
-
-            //TODO: What BasicConsume returns?
-            ChannelAccessor.Channel.BasicConsume(
-                queue: queueName,
-                autoAck: false,
-                consumer: Consumer
-            );
-        }
-
-        private void MessageReceived(object sender, BasicDeliverEventArgs e)
-        {
-            var context = new JobExecutionContext(typeof(TJob), Serializer.Deserialize(e.Body, ArgsType));
-            JobExecuter.Execute(context);
-            //TODO: How to ACK on success or Reject on failure?
-        }
-
-        public void Stop()
-        {
-            ChannelAccessor.Dispose();
-        }
-    }
-}

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

@@ -0,0 +1,153 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Options;
+using Nito.AsyncEx;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using Volo.Abp.RabbitMQ;
+
+namespace Volo.Abp.BackgroundJobs.RabbitMQ
+{
+    public class JobQueue<TArgs> : IJobQueue<TArgs>
+    {
+        protected Type JobType { get; }
+        protected string JobName { get; }
+        protected string QueueName { get; }
+
+        protected IChannelAccessor ChannelAccessor { get; private set; }
+        protected EventingBasicConsumer Consumer { get; private set; }
+
+        protected IChannelPool ChannelPool { get; }
+        protected AbpRabbitMqOptions RabbitMqOptions { get; }
+        protected BackgroundJobOptions BackgroundJobOptions { get; }
+        protected IRabbitMqSerializer Serializer { get; }
+        protected IBackgroundJobExecuter JobExecuter { get; }
+
+        protected AsyncLock SyncObj = new AsyncLock();
+        protected bool IsDiposed { get; private set; }
+
+        public JobQueue(
+            IChannelPool channelPool,
+            IRabbitMqSerializer serializer,
+            IOptions<AbpRabbitMqOptions> options,
+            IBackgroundJobExecuter jobExecuter,
+            IOptions<BackgroundJobOptions> backgroundJobOptions)
+        {
+            Serializer = serializer;
+            JobExecuter = jobExecuter;
+            BackgroundJobOptions = backgroundJobOptions.Value;
+            ChannelPool = channelPool;
+            RabbitMqOptions = options.Value;
+
+            JobName = BackgroundJobNameAttribute.GetName<TArgs>();
+            JobType = BackgroundJobOptions.GetJobType(JobName);
+            QueueName = "BackgroundJobs." + JobName; //TODO: Make prefix optional
+        }
+
+        public virtual async Task<string> Enqueue(TArgs args)
+        {
+            CheckDisposed();
+
+            using (await SyncObj.LockAsync())
+            {
+                await EnsureInitializedAsync();
+
+                await PublishAsync(args);
+
+                return null;
+            }
+        }
+
+        public Task StartAsync()
+        {
+            if (!BackgroundJobOptions.IsJobExecutionEnabled)
+            {
+                return Task.CompletedTask;
+            }
+
+            return EnsureInitializedAsync();
+        }
+
+        public virtual void Dispose()
+        {
+            if (IsDiposed)
+            {
+                return;
+            }
+
+            IsDiposed = true;
+
+            ChannelAccessor?.Dispose();
+        }
+
+        public virtual Task EnsureInitializedAsync()
+        {
+            if (ChannelAccessor != null)
+            {
+                return Task.CompletedTask;
+            }
+
+            ChannelAccessor = ChannelPool.Acquire(QueueName);
+
+            var queueOptions = RabbitMqOptions.Queues.GetOrDefault(QueueName)
+                               ?? new QueueOptions(QueueName);
+
+            queueOptions.Declare(ChannelAccessor.Channel);
+
+            if (BackgroundJobOptions.IsJobExecutionEnabled)
+            {
+                Consumer = new EventingBasicConsumer(ChannelAccessor.Channel);
+                Consumer.Received += MessageReceived;
+
+                //TODO: What BasicConsume returns?
+                ChannelAccessor.Channel.BasicConsume(
+                    queue: QueueName,
+                    autoAck: false,
+                    consumer: Consumer
+                );
+            }
+
+            return Task.CompletedTask;
+        }
+
+        protected virtual Task PublishAsync(TArgs args)
+        {
+            ChannelAccessor.Channel.BasicPublish(
+                exchange: "",
+                routingKey: QueueName,
+                basicProperties: CreateBasicPropertiesToPublish(),
+                body: Serializer.Serialize(args)
+            );
+
+            return Task.CompletedTask;
+        }
+
+        protected virtual IBasicProperties CreateBasicPropertiesToPublish()
+        {
+            var properties = ChannelAccessor.Channel.CreateBasicProperties();
+            properties.Persistent = true;
+            return properties;
+        }
+
+        protected virtual void MessageReceived(object sender, BasicDeliverEventArgs ea)
+        {
+            var context = new JobExecutionContext(
+                JobType,
+                Serializer.Deserialize(ea.Body, typeof(TArgs))
+            );
+
+            JobExecuter.Execute(context);
+
+            //TODO: How to ACK on success or Reject on failure?
+        }
+
+        private void CheckDisposed()
+        {
+            if (IsDiposed)
+            {
+                throw new AbpException("This object is disposed!");
+            }
+        }
+    }
+}

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

@@ -17,35 +17,12 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
             ChannelPool = channelPool;
         }
 
-        public Task<string> EnqueueAsync<TArgs>(TArgs args, BackgroundJobPriority priority = BackgroundJobPriority.Normal,
-            TimeSpan? delay = null)
-        {
-            var jobName = BackgroundJobNameAttribute.GetName<TArgs>();
-            var queueName = "BackgroundJobs." + jobName; //TODO: Make prefix optional
-
-            using (var channelAccessor = ChannelPool.Acquire(queueName))
-            {
-                var properties = channelAccessor.Channel.CreateBasicProperties();
-                properties.Persistent = true;
-
-                Publish(channelAccessor.Channel, queueName, args, properties);
-            }
-
-            return null; //TODO: Can RabbitMQ return a message identifier?
-        }
-
-        private void Publish<TArgs>(
-            IModel channel, 
-            string queueName, 
+        public Task<string> EnqueueAsync<TArgs>(
             TArgs args, 
-            IBasicProperties properties)
+            BackgroundJobPriority priority = BackgroundJobPriority.Normal,
+            TimeSpan? delay = null)
         {
-            channel.BasicPublish(
-                exchange: "",
-                routingKey: queueName,
-                basicProperties: properties,
-                body: Serializer.Serialize(args)
-            );
+            
         }
     }
 }

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

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

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

@@ -1,5 +1,6 @@
 using System.Collections.Generic;
 using JetBrains.Annotations;
+using RabbitMQ.Client;
 
 namespace Volo.Abp.RabbitMQ
 {
@@ -16,7 +17,11 @@ namespace Volo.Abp.RabbitMQ
 
         public IDictionary<string, object> Arguments { get; }
 
-        public QueueOptions([NotNull] string name, bool durable = true, bool exclusive = false, bool autoDelete = false)
+        public QueueOptions(
+            [NotNull] string name, 
+            bool durable = true, 
+            bool exclusive = false, 
+            bool autoDelete = false)
         {
             Name = name;
             Durable = durable;
@@ -24,5 +29,16 @@ namespace Volo.Abp.RabbitMQ
             AutoDelete = autoDelete;
             Arguments = new Dictionary<string, object>();
         }
+
+        public QueueDeclareOk Declare(IModel channel)
+        {
+            return channel.QueueDeclare(
+                queue: Name,
+                durable: Durable,
+                exclusive: Exclusive,
+                autoDelete: AutoDelete,
+                arguments: Arguments
+            );
+        }
     }
 }

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

@@ -2,7 +2,7 @@
 
 namespace Volo.Abp.RabbitMQ
 {
-    public class QueueDictionary : Dictionary<string, QueueOptions>
+    public class QueueOptionsDictionary : Dictionary<string, QueueOptions>
     {
 
     }