Browse Source

Make IJobQueueManager.Get method async to reduce AsyncHelper usage

pull/2422/head
Halil İbrahim Kalkan 7 years ago
parent
commit
647dd63e7e
  1. 5
      framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobQueueManager.cs
  2. 31
      framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueManager.cs
  3. 7
      framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/RabbitMqBackgroundJobManager.cs

5
framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobQueueManager.cs

@ -1,9 +1,10 @@
using Volo.Abp.Threading; using System.Threading.Tasks;
using Volo.Abp.Threading;
namespace Volo.Abp.BackgroundJobs.RabbitMQ namespace Volo.Abp.BackgroundJobs.RabbitMQ
{ {
public interface IJobQueueManager : IRunnable public interface IJobQueueManager : IRunnable
{ {
IJobQueue<TArgs> Get<TArgs>(); Task<IJobQueue<TArgs>> GetAsync<TArgs>();
} }
} }

31
framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueManager.cs

@ -4,6 +4,7 @@ using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using Nito.AsyncEx;
using Volo.Abp.DependencyInjection; using Volo.Abp.DependencyInjection;
using Volo.Abp.Threading; using Volo.Abp.Threading;
@ -17,6 +18,8 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
protected AbpBackgroundJobOptions Options { get; } protected AbpBackgroundJobOptions Options { get; }
protected SemaphoreSlim SyncSemaphore { get; }
public JobQueueManager( public JobQueueManager(
IOptions<AbpBackgroundJobOptions> options, IOptions<AbpBackgroundJobOptions> options,
IServiceProvider serviceProvider) IServiceProvider serviceProvider)
@ -24,6 +27,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
ServiceProvider = serviceProvider; ServiceProvider = serviceProvider;
Options = options.Value; Options = options.Value;
JobQueues = new ConcurrentDictionary<string, IRunnable>(); JobQueues = new ConcurrentDictionary<string, IRunnable>();
SyncSemaphore = new SemaphoreSlim(1, 1);
} }
public async Task StartAsync(CancellationToken cancellationToken = default) public async Task StartAsync(CancellationToken cancellationToken = default)
@ -51,20 +55,31 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
JobQueues.Clear(); JobQueues.Clear();
} }
public IJobQueue<TArgs> Get<TArgs>() public async Task<IJobQueue<TArgs>> GetAsync<TArgs>()
{ {
var jobConfiguration = Options.GetJob(typeof(TArgs)); var jobConfiguration = Options.GetJob(typeof(TArgs));
return (IJobQueue<TArgs>)JobQueues.GetOrAdd(jobConfiguration.JobName, _ => if (JobQueues.TryGetValue(jobConfiguration.JobName, out var jobQueue))
{
return (IJobQueue<TArgs>)jobQueue;
}
using (await SyncSemaphore.LockAsync())
{ {
var jobQueue = (IRunnable) ServiceProvider if (JobQueues.TryGetValue(jobConfiguration.JobName, out jobQueue))
.GetRequiredService(typeof(IJobQueue<>) {
.MakeGenericType(typeof(TArgs))); return (IJobQueue<TArgs>)jobQueue;
}
AsyncHelper.RunSync(() => jobQueue.StartAsync()); jobQueue = (IJobQueue<TArgs>)ServiceProvider
.GetRequiredService(typeof(IJobQueue<>).MakeGenericType(typeof(TArgs)));
return jobQueue; await jobQueue.StartAsync();
});
JobQueues.TryAdd(jobConfiguration.JobName, jobQueue);
return (IJobQueue<TArgs>)jobQueue;
}
} }
} }
} }

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

@ -14,14 +14,13 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
_jobQueueManager = jobQueueManager; _jobQueueManager = jobQueueManager;
} }
public Task<string> EnqueueAsync<TArgs>( public async Task<string> EnqueueAsync<TArgs>(
TArgs args, TArgs args,
BackgroundJobPriority priority = BackgroundJobPriority.Normal, BackgroundJobPriority priority = BackgroundJobPriority.Normal,
TimeSpan? delay = null) TimeSpan? delay = null)
{ {
return _jobQueueManager var jobQueue = await _jobQueueManager.GetAsync<TArgs>();
.Get<TArgs>() return await jobQueue.EnqueueAsync(args, priority, delay);
.EnqueueAsync(args, priority, delay);
} }
} }
} }

Loading…
Cancel
Save