mirror of https://github.com/abpframework/abp.git
34 changed files with 419 additions and 804 deletions
@ -1,62 +0,0 @@ |
|||||
using System; |
|
||||
using System.Collections.Concurrent; |
|
||||
using System.Threading; |
|
||||
using System.Threading.Tasks; |
|
||||
using Microsoft.Extensions.Options; |
|
||||
using Volo.Abp.DependencyInjection; |
|
||||
|
|
||||
namespace Volo.Abp.BackgroundJobs; |
|
||||
|
|
||||
public class AnonymousJobHandlerRegistry : IAnonymousJobHandlerRegistry, ISingletonDependency |
|
||||
{ |
|
||||
private readonly ConcurrentDictionary<string, Func<AnonymousJobExecutionContext, CancellationToken, Task>> _handlers = new(); |
|
||||
private readonly AbpBackgroundJobOptions _options; |
|
||||
|
|
||||
public AnonymousJobHandlerRegistry(IOptions<AbpBackgroundJobOptions> options) |
|
||||
{ |
|
||||
_options = options.Value; |
|
||||
} |
|
||||
|
|
||||
public virtual void Register(string jobName, Func<AnonymousJobExecutionContext, CancellationToken, Task> handler) |
|
||||
{ |
|
||||
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
|
||||
Check.NotNull(handler, nameof(handler)); |
|
||||
|
|
||||
_handlers[jobName] = handler; |
|
||||
} |
|
||||
|
|
||||
public virtual void Register(string jobName, Action<AnonymousJobExecutionContext, CancellationToken> handler) |
|
||||
{ |
|
||||
Check.NotNull(handler, nameof(handler)); |
|
||||
|
|
||||
Register(jobName, (context, ct) => |
|
||||
{ |
|
||||
handler(context, ct); |
|
||||
return Task.CompletedTask; |
|
||||
}); |
|
||||
} |
|
||||
|
|
||||
public virtual bool Unregister(string jobName) |
|
||||
{ |
|
||||
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
|
||||
return _handlers.TryRemove(jobName, out _); |
|
||||
} |
|
||||
|
|
||||
public virtual bool IsRegistered(string jobName) |
|
||||
{ |
|
||||
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
|
||||
return _handlers.ContainsKey(jobName) || _options.IsAnonymousJobRegistered(jobName); |
|
||||
} |
|
||||
|
|
||||
public virtual Func<AnonymousJobExecutionContext, CancellationToken, Task>? Get(string jobName) |
|
||||
{ |
|
||||
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
|
||||
|
|
||||
if (_handlers.TryGetValue(jobName, out var handler)) |
|
||||
{ |
|
||||
return handler; |
|
||||
} |
|
||||
|
|
||||
return _options.TryGetAnonymousHandler(jobName, out handler) ? handler : null; |
|
||||
} |
|
||||
} |
|
||||
@ -0,0 +1,112 @@ |
|||||
|
using System; |
||||
|
using System.Collections.Concurrent; |
||||
|
using System.Reflection; |
||||
|
using System.Threading.Tasks; |
||||
|
using Microsoft.Extensions.Options; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
using Volo.Abp.Json; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundJobs; |
||||
|
|
||||
|
public class DefaultDynamicBackgroundJobManager : IDynamicBackgroundJobManager, ITransientDependency |
||||
|
{ |
||||
|
private static readonly ConcurrentDictionary<Type, MethodInfo> EnqueueMethodCache = new(); |
||||
|
|
||||
|
protected IBackgroundJobManager BackgroundJobManager { get; } |
||||
|
protected IDynamicBackgroundJobHandlerRegistry HandlerRegistry { get; } |
||||
|
protected AbpBackgroundJobOptions BackgroundJobOptions { get; } |
||||
|
protected IJsonSerializer JsonSerializer { get; } |
||||
|
|
||||
|
public DefaultDynamicBackgroundJobManager( |
||||
|
IBackgroundJobManager backgroundJobManager, |
||||
|
IDynamicBackgroundJobHandlerRegistry handlerRegistry, |
||||
|
IOptions<AbpBackgroundJobOptions> backgroundJobOptions, |
||||
|
IJsonSerializer jsonSerializer) |
||||
|
{ |
||||
|
BackgroundJobManager = backgroundJobManager; |
||||
|
HandlerRegistry = handlerRegistry; |
||||
|
BackgroundJobOptions = backgroundJobOptions.Value; |
||||
|
JsonSerializer = jsonSerializer; |
||||
|
} |
||||
|
|
||||
|
public virtual async Task<string> EnqueueAsync( |
||||
|
string jobName, |
||||
|
object args, |
||||
|
BackgroundJobPriority priority = BackgroundJobPriority.Normal, |
||||
|
TimeSpan? delay = null) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
||||
|
Check.NotNull(args, nameof(args)); |
||||
|
|
||||
|
var jobConfiguration = BackgroundJobOptions.GetJobOrNull(jobName); |
||||
|
if (jobConfiguration != null) |
||||
|
{ |
||||
|
return await EnqueueTypedJobAsync(jobConfiguration, args, priority, delay); |
||||
|
} |
||||
|
|
||||
|
if (HandlerRegistry.IsRegistered(jobName)) |
||||
|
{ |
||||
|
return await EnqueueAnonymousJobAsync(jobName, args, priority, delay); |
||||
|
} |
||||
|
|
||||
|
throw new AbpException( |
||||
|
$"No typed job configuration or dynamic handler registered for job name: {jobName}"); |
||||
|
} |
||||
|
|
||||
|
public virtual void RegisterHandler( |
||||
|
string jobName, |
||||
|
DynamicBackgroundJobHandler handler) |
||||
|
{ |
||||
|
HandlerRegistry.Register(jobName, handler); |
||||
|
} |
||||
|
|
||||
|
public virtual bool UnregisterHandler(string jobName) |
||||
|
{ |
||||
|
return HandlerRegistry.Unregister(jobName); |
||||
|
} |
||||
|
|
||||
|
public virtual bool IsHandlerRegistered(string jobName) |
||||
|
{ |
||||
|
return HandlerRegistry.IsRegistered(jobName); |
||||
|
} |
||||
|
|
||||
|
protected virtual async Task<string> EnqueueTypedJobAsync( |
||||
|
BackgroundJobConfiguration jobConfiguration, |
||||
|
object args, |
||||
|
BackgroundJobPriority priority, |
||||
|
TimeSpan? delay) |
||||
|
{ |
||||
|
var argsType = jobConfiguration.ArgsType; |
||||
|
|
||||
|
// Normalize args to the expected type via JSON round-trip
|
||||
|
var json = JsonSerializer.Serialize(args); |
||||
|
var typedArgs = JsonSerializer.Deserialize(argsType, json); |
||||
|
|
||||
|
var enqueueMethod = GetOrCreateEnqueueMethod(argsType); |
||||
|
var task = (Task<string>)enqueueMethod.Invoke(BackgroundJobManager, [typedArgs, priority, delay])!; |
||||
|
return await task; |
||||
|
} |
||||
|
|
||||
|
protected virtual Task<string> EnqueueAnonymousJobAsync( |
||||
|
string jobName, |
||||
|
object args, |
||||
|
BackgroundJobPriority priority, |
||||
|
TimeSpan? delay) |
||||
|
{ |
||||
|
var jsonData = JsonSerializer.Serialize(args); |
||||
|
var anonymousArgs = new AnonymousJobArgs(jobName, jsonData); |
||||
|
return BackgroundJobManager.EnqueueAsync(anonymousArgs, priority, delay); |
||||
|
} |
||||
|
|
||||
|
private static MethodInfo GetOrCreateEnqueueMethod(Type argsType) |
||||
|
{ |
||||
|
return EnqueueMethodCache.GetOrAdd(argsType, static type => |
||||
|
{ |
||||
|
var method = typeof(IBackgroundJobManager).GetMethod( |
||||
|
nameof(IBackgroundJobManager.EnqueueAsync), |
||||
|
BindingFlags.Public | BindingFlags.Instance); |
||||
|
|
||||
|
return method!.MakeGenericMethod(type); |
||||
|
}); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,9 @@ |
|||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundJobs; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Represents a handler delegate for dynamic background jobs.
|
||||
|
/// </summary>
|
||||
|
public delegate Task DynamicBackgroundJobHandler(DynamicBackgroundJobExecutionContext context, CancellationToken cancellationToken); |
||||
@ -0,0 +1,40 @@ |
|||||
|
using System.Collections.Concurrent; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundJobs; |
||||
|
|
||||
|
public class DynamicBackgroundJobHandlerRegistry : IDynamicBackgroundJobHandlerRegistry, ISingletonDependency |
||||
|
{ |
||||
|
protected ConcurrentDictionary<string, DynamicBackgroundJobHandler> Handlers { get; } |
||||
|
|
||||
|
public DynamicBackgroundJobHandlerRegistry() |
||||
|
{ |
||||
|
Handlers = new ConcurrentDictionary<string, DynamicBackgroundJobHandler>(); |
||||
|
} |
||||
|
|
||||
|
public virtual void Register(string jobName, DynamicBackgroundJobHandler handler) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
||||
|
Check.NotNull(handler, nameof(handler)); |
||||
|
|
||||
|
Handlers[jobName] = handler; |
||||
|
} |
||||
|
|
||||
|
public virtual bool Unregister(string jobName) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
||||
|
return Handlers.TryRemove(jobName, out _); |
||||
|
} |
||||
|
|
||||
|
public virtual bool IsRegistered(string jobName) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
||||
|
return Handlers.ContainsKey(jobName); |
||||
|
} |
||||
|
|
||||
|
public virtual DynamicBackgroundJobHandler? Get(string jobName) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(jobName, nameof(jobName)); |
||||
|
return Handlers.TryGetValue(jobName, out var handler) ? handler : null; |
||||
|
} |
||||
|
} |
||||
@ -1,17 +0,0 @@ |
|||||
using System; |
|
||||
using System.Threading.Tasks; |
|
||||
|
|
||||
namespace Volo.Abp.BackgroundJobs; |
|
||||
|
|
||||
public interface IAnonymousJobHandlerRegistry |
|
||||
{ |
|
||||
void Register(string jobName, Func<AnonymousJobExecutionContext, System.Threading.CancellationToken, Task> handler); |
|
||||
|
|
||||
void Register(string jobName, Action<AnonymousJobExecutionContext, System.Threading.CancellationToken> handler); |
|
||||
|
|
||||
bool Unregister(string jobName); |
|
||||
|
|
||||
bool IsRegistered(string jobName); |
|
||||
|
|
||||
Func<AnonymousJobExecutionContext, System.Threading.CancellationToken, Task>? Get(string jobName); |
|
||||
} |
|
||||
@ -0,0 +1,12 @@ |
|||||
|
namespace Volo.Abp.BackgroundJobs; |
||||
|
|
||||
|
public interface IDynamicBackgroundJobHandlerRegistry |
||||
|
{ |
||||
|
void Register(string jobName, DynamicBackgroundJobHandler handler); |
||||
|
|
||||
|
bool Unregister(string jobName); |
||||
|
|
||||
|
bool IsRegistered(string jobName); |
||||
|
|
||||
|
DynamicBackgroundJobHandler? Get(string jobName); |
||||
|
} |
||||
@ -0,0 +1,51 @@ |
|||||
|
using System; |
||||
|
using System.Threading.Tasks; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundJobs; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Manages dynamic background jobs that can be enqueued by name
|
||||
|
/// without requiring a strongly-typed job args class at compile time.
|
||||
|
/// Also supports registering dynamic job handlers at runtime.
|
||||
|
/// </summary>
|
||||
|
public interface IDynamicBackgroundJobManager |
||||
|
{ |
||||
|
/// <summary>
|
||||
|
/// Enqueues a job by name with dynamic payload.
|
||||
|
/// If a typed job configuration exists for this name, the args will be
|
||||
|
/// deserialized to the configured args type and enqueued through the typed pipeline.
|
||||
|
/// If a dynamic handler is registered, the args will be wrapped as
|
||||
|
/// <see cref="AnonymousJobArgs"/> and enqueued through the standard typed pipeline.
|
||||
|
/// </summary>
|
||||
|
/// <param name="jobName">Name of the background job.</param>
|
||||
|
/// <param name="args">Job arguments (will be serialized to JSON).</param>
|
||||
|
/// <param name="priority">Job priority.</param>
|
||||
|
/// <param name="delay">Job delay (wait duration before first try).</param>
|
||||
|
/// <returns>Unique identifier of a background job.</returns>
|
||||
|
Task<string> EnqueueAsync( |
||||
|
string jobName, |
||||
|
object args, |
||||
|
BackgroundJobPriority priority = BackgroundJobPriority.Normal, |
||||
|
TimeSpan? delay = null); |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Registers a dynamic job handler at runtime.
|
||||
|
/// </summary>
|
||||
|
/// <param name="jobName">Unique name for the dynamic job.</param>
|
||||
|
/// <param name="handler">The handler delegate to execute when the job runs.</param>
|
||||
|
void RegisterHandler(string jobName, DynamicBackgroundJobHandler handler); |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Unregisters a previously registered dynamic job handler.
|
||||
|
/// </summary>
|
||||
|
/// <param name="jobName">Name of the dynamic job to unregister.</param>
|
||||
|
/// <returns>True if the handler was found and removed; false otherwise.</returns>
|
||||
|
bool UnregisterHandler(string jobName); |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Checks whether a dynamic handler is registered for the given job name.
|
||||
|
/// </summary>
|
||||
|
/// <param name="jobName">Name of the dynamic job.</param>
|
||||
|
/// <returns>True if registered; false otherwise.</returns>
|
||||
|
bool IsHandlerRegistered(string jobName); |
||||
|
} |
||||
@ -1,70 +1,25 @@ |
|||||
using System; |
using System; |
||||
using System.Threading.Tasks; |
using System.Threading.Tasks; |
||||
using Microsoft.Extensions.Logging; |
|
||||
using Microsoft.Extensions.Logging.Abstractions; |
|
||||
using Microsoft.Extensions.Options; |
|
||||
using Volo.Abp.DependencyInjection; |
using Volo.Abp.DependencyInjection; |
||||
using Volo.Abp.Json; |
|
||||
|
|
||||
namespace Volo.Abp.BackgroundJobs.RabbitMQ; |
namespace Volo.Abp.BackgroundJobs.RabbitMQ; |
||||
|
|
||||
[Dependency(ReplaceServices = true)] |
[Dependency(ReplaceServices = true)] |
||||
public class RabbitMqBackgroundJobManager : IBackgroundJobManager, ITransientDependency |
public class RabbitMqBackgroundJobManager : IBackgroundJobManager, ITransientDependency |
||||
{ |
{ |
||||
protected IJobQueueManager JobQueueManager { get; } |
private readonly IJobQueueManager _jobQueueManager; |
||||
protected IAnonymousJobHandlerRegistry AnonymousJobHandlerRegistry { get; } |
|
||||
protected AbpBackgroundJobOptions BackgroundJobOptions { get; } |
|
||||
protected IJsonSerializer JsonSerializer { get; } |
|
||||
public ILogger<RabbitMqBackgroundJobManager> Logger { get; set; } |
|
||||
|
|
||||
public RabbitMqBackgroundJobManager( |
public RabbitMqBackgroundJobManager(IJobQueueManager jobQueueManager) |
||||
IJobQueueManager jobQueueManager, |
|
||||
IAnonymousJobHandlerRegistry anonymousJobHandlerRegistry, |
|
||||
IOptions<AbpBackgroundJobOptions> backgroundJobOptions, |
|
||||
IJsonSerializer jsonSerializer) |
|
||||
{ |
{ |
||||
JobQueueManager = jobQueueManager; |
_jobQueueManager = jobQueueManager; |
||||
AnonymousJobHandlerRegistry = anonymousJobHandlerRegistry; |
|
||||
BackgroundJobOptions = backgroundJobOptions.Value; |
|
||||
JsonSerializer = jsonSerializer; |
|
||||
Logger = NullLogger<RabbitMqBackgroundJobManager>.Instance; |
|
||||
} |
} |
||||
|
|
||||
public virtual async 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) |
||||
{ |
{ |
||||
var jobQueue = await JobQueueManager.GetAsync<TArgs>(); |
var jobQueue = await _jobQueueManager.GetAsync<TArgs>(); |
||||
return (await jobQueue.EnqueueAsync(args, priority, delay))!; |
return (await jobQueue.EnqueueAsync(args, priority, delay))!; |
||||
} |
} |
||||
|
|
||||
public virtual async Task<string> EnqueueAsync( |
|
||||
string jobName, |
|
||||
object args, |
|
||||
BackgroundJobPriority priority = BackgroundJobPriority.Normal, |
|
||||
TimeSpan? delay = null) |
|
||||
{ |
|
||||
if (ShouldWrapAsAnonymousJob(jobName)) |
|
||||
{ |
|
||||
Logger.LogInformation( |
|
||||
"Wrapping job into anonymous transport. TransportJobName: {TransportJobName}, EffectiveJobName: {EffectiveJobName}", |
|
||||
AnonymousJobArgs.JobNameConstant, |
|
||||
jobName |
|
||||
); |
|
||||
var jsonData = JsonSerializer.Serialize(args); |
|
||||
var anonymousArgs = new AnonymousJobArgs(jobName, jsonData); |
|
||||
return await EnqueueAsync(AnonymousJobArgs.JobNameConstant, anonymousArgs, priority, delay); |
|
||||
} |
|
||||
|
|
||||
var jobQueue = await JobQueueManager.GetAsync(jobName); |
|
||||
return (await jobQueue.EnqueueAsync(args, priority, delay))!; |
|
||||
} |
|
||||
|
|
||||
protected virtual bool ShouldWrapAsAnonymousJob(string jobName) |
|
||||
{ |
|
||||
return jobName != AnonymousJobArgs.JobNameConstant && |
|
||||
AnonymousJobHandlerRegistry.IsRegistered(jobName) && |
|
||||
BackgroundJobOptions.GetJobOrNull(jobName) == null; |
|
||||
} |
|
||||
} |
} |
||||
|
|||||
Loading…
Reference in new issue