Browse Source

Refactor rabbitmq module.

pull/395/head
Halil ibrahim Kalkan 8 years ago
parent
commit
f99aa44687
  1. 12
      framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpBackgroundJobsRabbitMqModule.cs
  2. 2
      framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/IJobQueue.cs
  3. 42
      framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs
  4. 7
      framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/RabbitMqBackgroundJobManager.cs
  5. 5
      framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/Utf8JsonRabbitMqSerializer.cs

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

@ -20,13 +20,23 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
} }
public override void OnApplicationInitialization(ApplicationInitializationContext context) public override void OnApplicationInitialization(ApplicationInitializationContext context)
{
StartJobQueueManager(context);
}
public override void OnApplicationShutdown(ApplicationShutdownContext context)
{
StopJobQueueManager(context);
}
private static void StartJobQueueManager(ApplicationInitializationContext context)
{ {
context.ServiceProvider context.ServiceProvider
.GetRequiredService<IJobQueueManager>() .GetRequiredService<IJobQueueManager>()
.Start(); .Start();
} }
public override void OnApplicationShutdown(ApplicationShutdownContext context) private static void StopJobQueueManager(ApplicationShutdownContext context)
{ {
context.ServiceProvider context.ServiceProvider
.GetRequiredService<IJobQueueManager>() .GetRequiredService<IJobQueueManager>()

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

@ -4,7 +4,7 @@ using Volo.Abp.Threading;
namespace Volo.Abp.BackgroundJobs.RabbitMQ namespace Volo.Abp.BackgroundJobs.RabbitMQ
{ {
public interface IJobQueue<TArgs> : IRunnable, IDisposable public interface IJobQueue<in TArgs> : IRunnable, IDisposable
{ {
Task<string> EnqueueAsync( Task<string> EnqueueAsync(
TArgs args, TArgs args,

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

@ -12,56 +12,54 @@ using Volo.Abp.RabbitMQ;
namespace Volo.Abp.BackgroundJobs.RabbitMQ namespace Volo.Abp.BackgroundJobs.RabbitMQ
{ {
//TODO: Needs refactoring
public class JobQueue<TArgs> : IJobQueue<TArgs> public class JobQueue<TArgs> : IJobQueue<TArgs>
{ {
private const string ChannelPrefix = "JobQueue."; private const string ChannelPrefix = "JobQueue.";
protected BackgroundJobConfiguration JobConfiguration { get; } protected BackgroundJobConfiguration JobConfiguration { get; }
protected JobQueueConfiguration QueueConfiguration { get; } protected JobQueueConfiguration QueueConfiguration { get; }
protected IChannelAccessor ChannelAccessor { get; private set; } protected IChannelAccessor ChannelAccessor { get; private set; }
protected EventingBasicConsumer Consumer { get; private set; } protected EventingBasicConsumer Consumer { get; private set; }
public ILogger<JobQueue<TArgs>> Logger { get; set; } public ILogger<JobQueue<TArgs>> Logger { get; set; }
protected IChannelPool ChannelPool { get; }
protected AbpRabbitMqOptions RabbitMqOptions { get; }
protected BackgroundJobOptions BackgroundJobOptions { get; } protected BackgroundJobOptions BackgroundJobOptions { get; }
protected RabbitMqBackgroundJobOptions RabbitMqBackgroundJobOptions { get; }
protected IChannelPool ChannelPool { get; }
protected IRabbitMqSerializer Serializer { get; } protected IRabbitMqSerializer Serializer { get; }
protected IBackgroundJobExecuter JobExecuter { get; } protected IBackgroundJobExecuter JobExecuter { get; }
protected RabbitMqBackgroundJobOptions RabbitMqBackgroundJobOptions { get; }
protected AsyncLock SyncObj = new AsyncLock(); protected AsyncLock SyncObj = new AsyncLock();
protected bool IsDiposed { get; private set; } protected bool IsDiposed { get; private set; }
public JobQueue( public JobQueue(
IOptions<BackgroundJobOptions> backgroundJobOptions,
IOptions<RabbitMqBackgroundJobOptions> rabbitMqBackgroundJobOptions,
IChannelPool channelPool, IChannelPool channelPool,
IRabbitMqSerializer serializer, IRabbitMqSerializer serializer,
IOptions<AbpRabbitMqOptions> rabbitMqOptions, IBackgroundJobExecuter jobExecuter)
IBackgroundJobExecuter jobExecuter,
IOptions<BackgroundJobOptions> backgroundJobOptions,
IOptions<RabbitMqBackgroundJobOptions> rabbitMqBackgroundJobOptions)
{ {
BackgroundJobOptions = backgroundJobOptions.Value;
RabbitMqBackgroundJobOptions = rabbitMqBackgroundJobOptions.Value;
Serializer = serializer; Serializer = serializer;
JobExecuter = jobExecuter; JobExecuter = jobExecuter;
BackgroundJobOptions = backgroundJobOptions.Value;
ChannelPool = channelPool; ChannelPool = channelPool;
RabbitMqOptions = rabbitMqOptions.Value;
RabbitMqBackgroundJobOptions = rabbitMqBackgroundJobOptions.Value;
JobConfiguration = BackgroundJobOptions.GetJob(typeof(TArgs)); JobConfiguration = BackgroundJobOptions.GetJob(typeof(TArgs));
QueueConfiguration = GetOrCreateJobQueueConfiguration();
QueueConfiguration = RabbitMqBackgroundJobOptions.JobQueues.GetOrDefault(typeof(TArgs)) ??
new JobQueueConfiguration(
typeof(TArgs),
RabbitMqBackgroundJobOptions.DefaultQueueNamePrefix + JobConfiguration.JobName
);
Logger = NullLogger<JobQueue<TArgs>>.Instance; Logger = NullLogger<JobQueue<TArgs>>.Instance;
} }
protected virtual JobQueueConfiguration GetOrCreateJobQueueConfiguration()
{
return RabbitMqBackgroundJobOptions.JobQueues.GetOrDefault(typeof(TArgs)) ??
new JobQueueConfiguration(
typeof(TArgs),
RabbitMqBackgroundJobOptions.DefaultQueueNamePrefix + JobConfiguration.JobName
);
}
public virtual async Task<string> EnqueueAsync( public virtual async Task<string> EnqueueAsync(
TArgs args, TArgs args,
BackgroundJobPriority priority = BackgroundJobPriority.Normal, BackgroundJobPriority priority = BackgroundJobPriority.Normal,
@ -79,7 +77,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
} }
} }
public async Task StartAsync(CancellationToken cancellationToken = default) public virtual async Task StartAsync(CancellationToken cancellationToken = default)
{ {
CheckDisposed(); CheckDisposed();
@ -94,7 +92,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
} }
} }
public Task StopAsync(CancellationToken cancellationToken = default) public virtual Task StopAsync(CancellationToken cancellationToken = default)
{ {
Dispose(); Dispose();
return Task.CompletedTask; return Task.CompletedTask;
@ -191,7 +189,7 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
} }
} }
private void CheckDisposed() protected void CheckDisposed()
{ {
if (IsDiposed) if (IsDiposed)
{ {

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

@ -1,8 +1,6 @@
using System; using System;
using System.Threading.Tasks; using System.Threading.Tasks;
using RabbitMQ.Client;
using Volo.Abp.DependencyInjection; using Volo.Abp.DependencyInjection;
using Volo.Abp.RabbitMQ;
namespace Volo.Abp.BackgroundJobs.RabbitMQ namespace Volo.Abp.BackgroundJobs.RabbitMQ
{ {
@ -20,8 +18,9 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
BackgroundJobPriority priority = BackgroundJobPriority.Normal, BackgroundJobPriority priority = BackgroundJobPriority.Normal,
TimeSpan? delay = null) TimeSpan? delay = null)
{ {
var jobQueue = _jobQueueManager.Get<TArgs>(); return _jobQueueManager
return jobQueue.EnqueueAsync(args, priority, delay); .Get<TArgs>()
.EnqueueAsync(args, priority, delay);
} }
} }
} }

5
framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/Utf8JsonRabbitMqSerializer.cs

@ -23,10 +23,5 @@ namespace Volo.Abp.RabbitMQ
{ {
return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value)); return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value));
} }
public T Deserialize<T>(string value)
{
return _jsonSerializer.Deserialize<T>(value);
}
} }
} }
Loading…
Cancel
Save