mirror of https://github.com/abpframework/abp.git
Browse Source
Introduce runtime dynamic background workers: add DynamicBackgroundWorkerExecutionContext, DynamicBackgroundWorkerSchedule, IDynamicBackgroundWorkerHandlerRegistry and its implementation. Extend IBackgroundWorkerManager with AddAsync overloads to register handlers by name and schedule. Provide InMemoryDynamicBackgroundWorker for in-process execution and provider-specific adapters/implementations for Hangfire, Quartz and TickerQ (including Hangfire/Quartz/TickerQ adapters and manager changes) to schedule and execute dynamic handlers. Update BackgroundWorkerManager to hold IServiceProvider and the handler registry and wire DI through constructors. Add a docs example and unit tests to verify handler registration and execution.pull/25066/head
21 changed files with 590 additions and 14 deletions
@ -0,0 +1,31 @@ |
|||||
|
using System; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundWorkers.Hangfire; |
||||
|
|
||||
|
public class HangfireDynamicBackgroundWorkerAdapter : ITransientDependency |
||||
|
{ |
||||
|
protected IDynamicBackgroundWorkerHandlerRegistry DynamicBackgroundWorkerHandlerRegistry { get; } |
||||
|
protected IServiceProvider ServiceProvider { get; } |
||||
|
|
||||
|
public HangfireDynamicBackgroundWorkerAdapter( |
||||
|
IDynamicBackgroundWorkerHandlerRegistry dynamicBackgroundWorkerHandlerRegistry, |
||||
|
IServiceProvider serviceProvider) |
||||
|
{ |
||||
|
DynamicBackgroundWorkerHandlerRegistry = dynamicBackgroundWorkerHandlerRegistry; |
||||
|
ServiceProvider = serviceProvider; |
||||
|
} |
||||
|
|
||||
|
public virtual async Task DoWorkAsync(string workerName, CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
var handler = DynamicBackgroundWorkerHandlerRegistry.Get(workerName); |
||||
|
if (handler == null) |
||||
|
{ |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
await handler(new DynamicBackgroundWorkerExecutionContext(workerName, ServiceProvider), cancellationToken); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,41 @@ |
|||||
|
using System; |
||||
|
using System.Threading.Tasks; |
||||
|
using Quartz; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundWorkers.Quartz; |
||||
|
|
||||
|
public class QuartzDynamicBackgroundWorkerAdapter : IJob, ITransientDependency |
||||
|
{ |
||||
|
protected IDynamicBackgroundWorkerHandlerRegistry DynamicBackgroundWorkerHandlerRegistry { get; } |
||||
|
protected IServiceProvider ServiceProvider { get; } |
||||
|
|
||||
|
public QuartzDynamicBackgroundWorkerAdapter( |
||||
|
IDynamicBackgroundWorkerHandlerRegistry dynamicBackgroundWorkerHandlerRegistry, |
||||
|
IServiceProvider serviceProvider) |
||||
|
{ |
||||
|
DynamicBackgroundWorkerHandlerRegistry = dynamicBackgroundWorkerHandlerRegistry; |
||||
|
ServiceProvider = serviceProvider; |
||||
|
} |
||||
|
|
||||
|
public virtual async Task Execute(IJobExecutionContext context) |
||||
|
{ |
||||
|
var workerName = context.MergedJobDataMap.GetString(QuartzBackgroundWorkerManager.DynamicWorkerNameKey); |
||||
|
if (string.IsNullOrWhiteSpace(workerName)) |
||||
|
{ |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
var nonNullWorkerName = workerName!; |
||||
|
var handler = DynamicBackgroundWorkerHandlerRegistry.Get(nonNullWorkerName); |
||||
|
if (handler == null) |
||||
|
{ |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
await handler( |
||||
|
new DynamicBackgroundWorkerExecutionContext(nonNullWorkerName, ServiceProvider), |
||||
|
context.CancellationToken |
||||
|
); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,16 @@ |
|||||
|
using System; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundWorkers; |
||||
|
|
||||
|
public class DynamicBackgroundWorkerExecutionContext |
||||
|
{ |
||||
|
public string WorkerName { get; } |
||||
|
|
||||
|
public IServiceProvider ServiceProvider { get; } |
||||
|
|
||||
|
public DynamicBackgroundWorkerExecutionContext(string workerName, IServiceProvider serviceProvider) |
||||
|
{ |
||||
|
WorkerName = Check.NotNullOrWhiteSpace(workerName, nameof(workerName)); |
||||
|
ServiceProvider = Check.NotNull(serviceProvider, nameof(serviceProvider)); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,43 @@ |
|||||
|
using System; |
||||
|
using System.Collections.Concurrent; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundWorkers; |
||||
|
|
||||
|
public class DynamicBackgroundWorkerHandlerRegistry : IDynamicBackgroundWorkerHandlerRegistry, ISingletonDependency |
||||
|
{ |
||||
|
protected ConcurrentDictionary<string, Func<DynamicBackgroundWorkerExecutionContext, CancellationToken, Task>> Handlers { get; } |
||||
|
|
||||
|
public DynamicBackgroundWorkerHandlerRegistry() |
||||
|
{ |
||||
|
Handlers = new ConcurrentDictionary<string, Func<DynamicBackgroundWorkerExecutionContext, CancellationToken, Task>>(); |
||||
|
} |
||||
|
|
||||
|
public virtual void Register(string workerName, Func<DynamicBackgroundWorkerExecutionContext, CancellationToken, Task> handler) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(workerName, nameof(workerName)); |
||||
|
Check.NotNull(handler, nameof(handler)); |
||||
|
|
||||
|
Handlers[workerName] = handler; |
||||
|
} |
||||
|
|
||||
|
public virtual bool Unregister(string workerName) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(workerName, nameof(workerName)); |
||||
|
return Handlers.TryRemove(workerName, out _); |
||||
|
} |
||||
|
|
||||
|
public virtual bool IsRegistered(string workerName) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(workerName, nameof(workerName)); |
||||
|
return Handlers.ContainsKey(workerName); |
||||
|
} |
||||
|
|
||||
|
public virtual Func<DynamicBackgroundWorkerExecutionContext, CancellationToken, Task>? Get(string workerName) |
||||
|
{ |
||||
|
Check.NotNullOrWhiteSpace(workerName, nameof(workerName)); |
||||
|
return Handlers.TryGetValue(workerName, out var handler) ? handler : null; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,10 @@ |
|||||
|
namespace Volo.Abp.BackgroundWorkers; |
||||
|
|
||||
|
public class DynamicBackgroundWorkerSchedule |
||||
|
{ |
||||
|
public const int DefaultPeriod = 60000; |
||||
|
|
||||
|
public int? Period { get; set; } |
||||
|
|
||||
|
public string? CronExpression { get; set; } |
||||
|
} |
||||
@ -0,0 +1,16 @@ |
|||||
|
using System; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundWorkers; |
||||
|
|
||||
|
public interface IDynamicBackgroundWorkerHandlerRegistry |
||||
|
{ |
||||
|
void Register(string workerName, Func<DynamicBackgroundWorkerExecutionContext, CancellationToken, Task> handler); |
||||
|
|
||||
|
bool Unregister(string workerName); |
||||
|
|
||||
|
bool IsRegistered(string workerName); |
||||
|
|
||||
|
Func<DynamicBackgroundWorkerExecutionContext, CancellationToken, Task>? Get(string workerName); |
||||
|
} |
||||
@ -0,0 +1,40 @@ |
|||||
|
using System.Threading.Tasks; |
||||
|
using Microsoft.Extensions.DependencyInjection; |
||||
|
using Microsoft.Extensions.Logging; |
||||
|
using Volo.Abp.Threading; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundWorkers; |
||||
|
|
||||
|
public class InMemoryDynamicBackgroundWorker : AsyncPeriodicBackgroundWorkerBase |
||||
|
{ |
||||
|
protected string WorkerName { get; } |
||||
|
protected IDynamicBackgroundWorkerHandlerRegistry HandlerRegistry { get; } |
||||
|
|
||||
|
public InMemoryDynamicBackgroundWorker( |
||||
|
string workerName, |
||||
|
DynamicBackgroundWorkerSchedule schedule, |
||||
|
AbpAsyncTimer timer, |
||||
|
IServiceScopeFactory serviceScopeFactory, |
||||
|
IDynamicBackgroundWorkerHandlerRegistry handlerRegistry) |
||||
|
: base(timer, serviceScopeFactory) |
||||
|
{ |
||||
|
WorkerName = Check.NotNullOrWhiteSpace(workerName, nameof(workerName)); |
||||
|
Check.NotNull(schedule, nameof(schedule)); |
||||
|
HandlerRegistry = Check.NotNull(handlerRegistry, nameof(handlerRegistry)); |
||||
|
|
||||
|
Timer.Period = schedule.Period ?? DynamicBackgroundWorkerSchedule.DefaultPeriod; |
||||
|
CronExpression = schedule.CronExpression; |
||||
|
} |
||||
|
|
||||
|
protected override async Task DoWorkAsync(PeriodicBackgroundWorkerContext workerContext) |
||||
|
{ |
||||
|
var handler = HandlerRegistry.Get(WorkerName); |
||||
|
if (handler == null) |
||||
|
{ |
||||
|
Logger.LogWarning("No dynamic background worker handler registered for: {WorkerName}", WorkerName); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
await handler(new DynamicBackgroundWorkerExecutionContext(WorkerName, workerContext.ServiceProvider), workerContext.CancellationToken); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,64 @@ |
|||||
|
using System; |
||||
|
using System.Threading.Tasks; |
||||
|
using Shouldly; |
||||
|
using Volo.Abp.BackgroundWorkers; |
||||
|
using Xunit; |
||||
|
|
||||
|
namespace Volo.Abp.BackgroundJobs; |
||||
|
|
||||
|
public class DynamicBackgroundWorkerManager_Tests : BackgroundJobsTestBase |
||||
|
{ |
||||
|
private readonly IBackgroundWorkerManager _backgroundWorkerManager; |
||||
|
private readonly IDynamicBackgroundWorkerHandlerRegistry _handlerRegistry; |
||||
|
|
||||
|
public DynamicBackgroundWorkerManager_Tests() |
||||
|
{ |
||||
|
_backgroundWorkerManager = GetRequiredService<IBackgroundWorkerManager>(); |
||||
|
_handlerRegistry = GetRequiredService<IDynamicBackgroundWorkerHandlerRegistry>(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Register_Dynamic_Handler_When_Added() |
||||
|
{ |
||||
|
var workerName = "dynamic-worker-" + Guid.NewGuid(); |
||||
|
|
||||
|
await _backgroundWorkerManager.AddAsync( |
||||
|
workerName, |
||||
|
new DynamicBackgroundWorkerSchedule |
||||
|
{ |
||||
|
Period = 1000 |
||||
|
}, |
||||
|
(_, _) => Task.CompletedTask |
||||
|
); |
||||
|
|
||||
|
_handlerRegistry.IsRegistered(workerName).ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Execute_Dynamic_Handler() |
||||
|
{ |
||||
|
var workerName = "dynamic-worker-" + Guid.NewGuid(); |
||||
|
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously); |
||||
|
|
||||
|
await _backgroundWorkerManager.AddAsync( |
||||
|
workerName, |
||||
|
new DynamicBackgroundWorkerSchedule |
||||
|
{ |
||||
|
Period = 50 |
||||
|
}, |
||||
|
(context, _) => |
||||
|
{ |
||||
|
if (context.WorkerName == workerName) |
||||
|
{ |
||||
|
tcs.TrySetResult(true); |
||||
|
} |
||||
|
|
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
); |
||||
|
|
||||
|
var completedTask = await Task.WhenAny(tcs.Task, Task.Delay(5000)); |
||||
|
completedTask.ShouldBe(tcs.Task); |
||||
|
(await tcs.Task).ShouldBeTrue(); |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue