From 8b1a9e4ebbc1b0e6fc31a5065b89e61dac6e593c Mon Sep 17 00:00:00 2001 From: Halil ibrahim Kalkan Date: Thu, 26 Jul 2018 09:26:47 +0300 Subject: [PATCH] Introcude JobQueue --- .../Volo.Abp.BackgroundJobs.RabbitMQ.csproj | 1 + .../AbpBackgroundJobsRabbitMqModule.cs | 2 +- .../BackgroundJobs/RabbitMQ/IJobListener.cs | 8 - .../Abp/BackgroundJobs/RabbitMQ/IJobQueue.cs | 12 ++ .../BackgroundJobs/RabbitMQ/JobListener.cs | 62 ------- .../Abp/BackgroundJobs/RabbitMQ/JobQueue.cs | 153 ++++++++++++++++++ .../RabbitMQ/RabbitMqBackgroundJobManager.cs | 31 +--- .../Volo/Abp/RabbitMQ/AbpRabbitMqOptions.cs | 4 +- .../Volo/Abp/RabbitMQ/QueueOptions.cs | 18 ++- ...ictionary.cs => QueueOptionsDictionary.cs} | 2 +- 10 files changed, 191 insertions(+), 102 deletions(-) delete mode 100644 framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobListener.cs create mode 100644 framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobQueue.cs delete mode 100644 framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobListener.cs create mode 100644 framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs rename framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/{QueueDictionary.cs => QueueOptionsDictionary.cs} (51%) diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo.Abp.BackgroundJobs.RabbitMQ.csproj b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo.Abp.BackgroundJobs.RabbitMQ.csproj index f29ed7d006..8ee2c499b6 100644 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo.Abp.BackgroundJobs.RabbitMQ.csproj +++ b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo.Abp.BackgroundJobs.RabbitMQ.csproj @@ -16,6 +16,7 @@ + diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpBackgroundJobsRabbitMqModule.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpBackgroundJobsRabbitMqModule.cs index 590afbef1b..863219e7ef 100644 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpBackgroundJobsRabbitMqModule.cs +++ b/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(); } diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobListener.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobListener.cs deleted file mode 100644 index 1bfb96fec6..0000000000 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobListener.cs +++ /dev/null @@ -1,8 +0,0 @@ -namespace Volo.Abp.BackgroundJobs.RabbitMQ -{ - public interface IJobListener - { - void Start(); - void Stop(); - } -} \ No newline at end of file diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobQueue.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobQueue.cs new file mode 100644 index 0000000000..40678e494c --- /dev/null +++ b/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 : IDisposable + { + Task Enqueue(TArgs args); + + Task StartAsync(); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobListener.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobListener.cs deleted file mode 100644 index 19252a03ad..0000000000 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobListener.cs +++ /dev/null @@ -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 : 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(); - } - } -} \ No newline at end of file diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs new file mode 100644 index 0000000000..2744ec7edb --- /dev/null +++ b/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 : IJobQueue + { + 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 options, + IBackgroundJobExecuter jobExecuter, + IOptions backgroundJobOptions) + { + Serializer = serializer; + JobExecuter = jobExecuter; + BackgroundJobOptions = backgroundJobOptions.Value; + ChannelPool = channelPool; + RabbitMqOptions = options.Value; + + JobName = BackgroundJobNameAttribute.GetName(); + JobType = BackgroundJobOptions.GetJobType(JobName); + QueueName = "BackgroundJobs." + JobName; //TODO: Make prefix optional + } + + public virtual async Task 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!"); + } + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/RabbitMqBackgroundJobManager.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/RabbitMqBackgroundJobManager.cs index 9892727d54..08ea5aaa7e 100644 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/RabbitMqBackgroundJobManager.cs +++ b/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 EnqueueAsync(TArgs args, BackgroundJobPriority priority = BackgroundJobPriority.Normal, - TimeSpan? delay = null) - { - var jobName = BackgroundJobNameAttribute.GetName(); - 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( - IModel channel, - string queueName, + public Task EnqueueAsync( TArgs args, - IBasicProperties properties) + BackgroundJobPriority priority = BackgroundJobPriority.Normal, + TimeSpan? delay = null) { - channel.BasicPublish( - exchange: "", - routingKey: queueName, - basicProperties: properties, - body: Serializer.Serialize(args) - ); + } } } diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqOptions.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqOptions.cs index 590aab97b7..b5d8c5df30 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqOptions.cs +++ b/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(); } } } diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueOptions.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueOptions.cs index 55905c9e3b..e5c2eda387 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueOptions.cs +++ b/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 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(); } + + public QueueDeclareOk Declare(IModel channel) + { + return channel.QueueDeclare( + queue: Name, + durable: Durable, + exclusive: Exclusive, + autoDelete: AutoDelete, + arguments: Arguments + ); + } } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDictionary.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueOptionsDictionary.cs similarity index 51% rename from framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDictionary.cs rename to framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueOptionsDictionary.cs index 9d0015d51c..056e4baefe 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDictionary.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueOptionsDictionary.cs @@ -2,7 +2,7 @@ namespace Volo.Abp.RabbitMQ { - public class QueueDictionary : Dictionary + public class QueueOptionsDictionary : Dictionary { }