Merge pull request #12420 from abpframework/CancellationTokenInWorker

Add `CancellationToken` paramter to `AddAsync` and `DoWorkAsync` methods.

framework/src/Volo.Abp.BackgroundWorkers.Hangfire/Volo/Abp/BackgroundWorkers/Hangfire/HangfireBackgroundWorkerBase.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace Volo.Abp.BackgroundWorkers.Hangfire;
@@ -8,12 +9,12 @@ public abstract class HangfireBackgroundWorkerBase : BackgroundWorkerBase, IHang
     public string RecurringJobId { get; set; }
 
     public string CronExpression { get; set; }
-    
+
     public TimeZoneInfo TimeZone { get; set; }
-    
+
     public string Queue { get; set; }
 
-    public abstract Task DoWorkAsync();
+    public abstract Task DoWorkAsync(CancellationToken cancellationToken = default);
 
     protected HangfireBackgroundWorkerBase()
     {

framework/src/Volo.Abp.BackgroundWorkers.Hangfire/Volo/Abp/BackgroundWorkers/Hangfire/HangfireBackgroundWorkerManager.cs

@@ -33,19 +33,19 @@ public class HangfireBackgroundWorkerManager : IBackgroundWorkerManager, ISingle
         return Task.CompletedTask;
     }
 
-    public Task AddAsync(IBackgroundWorker worker)
+    public Task AddAsync(IBackgroundWorker worker, CancellationToken cancellationToken = default)
     {
         if (worker is IHangfireBackgroundWorker hangfireBackgroundWorker)
         {
             var unProxyWorker = ProxyHelper.UnProxy(hangfireBackgroundWorker);
             if (hangfireBackgroundWorker.RecurringJobId.IsNullOrWhiteSpace())
             {
-                RecurringJob.AddOrUpdate(() => ((IHangfireBackgroundWorker)unProxyWorker).DoWorkAsync(),
+                RecurringJob.AddOrUpdate(() => ((IHangfireBackgroundWorker)unProxyWorker).DoWorkAsync(cancellationToken),
                     hangfireBackgroundWorker.CronExpression, hangfireBackgroundWorker.TimeZone, hangfireBackgroundWorker.Queue);
             }
             else
             {
-                RecurringJob.AddOrUpdate(hangfireBackgroundWorker.RecurringJobId, () => ((IHangfireBackgroundWorker)unProxyWorker).DoWorkAsync(),
+                RecurringJob.AddOrUpdate(hangfireBackgroundWorker.RecurringJobId, () => ((IHangfireBackgroundWorker)unProxyWorker).DoWorkAsync(cancellationToken),
                     hangfireBackgroundWorker.CronExpression, hangfireBackgroundWorker.TimeZone, hangfireBackgroundWorker.Queue);
             }
         }
@@ -80,7 +80,7 @@ public class HangfireBackgroundWorkerManager : IBackgroundWorkerManager, ISingle
             var adapterType = typeof(HangfirePeriodicBackgroundWorkerAdapter<>).MakeGenericType(ProxyHelper.GetUnProxiedType(worker));
             var workerAdapter = Activator.CreateInstance(adapterType) as IHangfireBackgroundWorker;
 
-            RecurringJob.AddOrUpdate(() => workerAdapter.DoWorkAsync(), GetCron(period.Value), workerAdapter.TimeZone, workerAdapter.Queue);
+            RecurringJob.AddOrUpdate(() => workerAdapter.DoWorkAsync(cancellationToken), GetCron(period.Value), workerAdapter.TimeZone, workerAdapter.Queue);
         }
 
         return Task.CompletedTask;

framework/src/Volo.Abp.BackgroundWorkers.Hangfire/Volo/Abp/BackgroundWorkers/Hangfire/HangfirePeriodicBackgroundWorkerAdapter.cs

@@ -1,4 +1,5 @@
 using System.Reflection;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 
@@ -17,9 +18,9 @@ public class HangfirePeriodicBackgroundWorkerAdapter<TWorker> : HangfireBackgrou
         _doWorkMethod = typeof(TWorker).GetMethod("DoWork", BindingFlags.Instance | BindingFlags.NonPublic);
     }
 
-    public async override Task DoWorkAsync()
+    public async override Task DoWorkAsync(CancellationToken cancellationToken = default)
     {
-        var workerContext = new PeriodicBackgroundWorkerContext(ServiceProvider);
+        var workerContext = new PeriodicBackgroundWorkerContext(ServiceProvider, cancellationToken);
         var worker = ServiceProvider.GetRequiredService<TWorker>();
 
         switch (worker)

framework/src/Volo.Abp.BackgroundWorkers.Hangfire/Volo/Abp/BackgroundWorkers/Hangfire/IHangfireBackgroundWorker.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace Volo.Abp.BackgroundWorkers.Hangfire;
@@ -8,12 +9,10 @@ public interface IHangfireBackgroundWorker : IBackgroundWorker
     string RecurringJobId { get; set; }
 
     string CronExpression { get; set; }
-    
+
     TimeZoneInfo TimeZone  { get; set; }
 
     string Queue  { get; set; }
 
-    Task DoWorkAsync();
+    Task DoWorkAsync(CancellationToken cancellationToken = default);
 }
-
-

framework/src/Volo.Abp.BackgroundWorkers.Quartz/Volo/Abp/BackgroundWorkers/Quartz/QuartzBackgroundWorkerManager.cs

@@ -4,7 +4,6 @@ using System.Threading.Tasks;
 using Quartz;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.DynamicProxy;
-using Volo.Abp.Threading;
 
 namespace Volo.Abp.BackgroundWorkers.Quartz;
 
@@ -34,12 +33,12 @@ public class QuartzBackgroundWorkerManager : IBackgroundWorkerManager, ISingleto
         }
     }
 
-    public virtual async Task AddAsync(IBackgroundWorker worker)
+    public virtual async Task AddAsync(IBackgroundWorker worker, CancellationToken cancellationToken = default)
     {
-        await ReScheduleJobAsync(worker);
+        await ReScheduleJobAsync(worker, cancellationToken);
     }
 
-    protected virtual async Task ReScheduleJobAsync(IBackgroundWorker worker)
+    protected virtual async Task ReScheduleJobAsync(IBackgroundWorker worker, CancellationToken cancellationToken = default)
     {
         if (worker is IQuartzBackgroundWorker quartzWork)
         {
@@ -52,7 +51,7 @@ public class QuartzBackgroundWorkerManager : IBackgroundWorkerManager, ISingleto
             }
             else
             {
-                await DefaultScheduleJobAsync(quartzWork);
+                await DefaultScheduleJobAsync(quartzWork, cancellationToken);
             }
         }
         else
@@ -65,22 +64,22 @@ public class QuartzBackgroundWorkerManager : IBackgroundWorkerManager, ISingleto
 
             if (workerAdapter?.Trigger != null)
             {
-                await DefaultScheduleJobAsync(workerAdapter);
+                await DefaultScheduleJobAsync(workerAdapter, cancellationToken);
             }
         }
     }
 
-    protected virtual async Task DefaultScheduleJobAsync(IQuartzBackgroundWorker quartzWork)
+    protected virtual async Task DefaultScheduleJobAsync(IQuartzBackgroundWorker quartzWork, CancellationToken cancellationToken = default)
     {
-        if (await _scheduler.CheckExists(quartzWork.JobDetail.Key))
+        if (await _scheduler.CheckExists(quartzWork.JobDetail.Key, cancellationToken))
         {
-            await _scheduler.AddJob(quartzWork.JobDetail, true, true);
-            await _scheduler.ResumeJob(quartzWork.JobDetail.Key);
-            await _scheduler.RescheduleJob(quartzWork.Trigger.Key, quartzWork.Trigger);
+            await _scheduler.AddJob(quartzWork.JobDetail, true, true, cancellationToken);
+            await _scheduler.ResumeJob(quartzWork.JobDetail.Key, cancellationToken);
+            await _scheduler.RescheduleJob(quartzWork.Trigger.Key, quartzWork.Trigger, cancellationToken);
         }
         else
         {
-            await _scheduler.ScheduleJob(quartzWork.JobDetail, quartzWork.Trigger);
+            await _scheduler.ScheduleJob(quartzWork.JobDetail, quartzWork.Trigger, cancellationToken);
         }
     }
 }

framework/src/Volo.Abp.BackgroundWorkers.Quartz/Volo/Abp/BackgroundWorkers/Quartz/QuartzPeriodicBackgroundWorkerAdapter.cs

@@ -67,10 +67,10 @@ public class QuartzPeriodicBackgroundWorkerAdapter<TWorker> : QuartzBackgroundWo
             .Build();
     }
 
-    public override async Task Execute(IJobExecutionContext context)
+    public async override Task Execute(IJobExecutionContext context)
     {
         var worker = (IBackgroundWorker) ServiceProvider.GetService(typeof(TWorker));
-        var workerContext = new PeriodicBackgroundWorkerContext(ServiceProvider);
+        var workerContext = new PeriodicBackgroundWorkerContext(ServiceProvider, context.CancellationToken);
 
         switch (worker)
         {

framework/src/Volo.Abp.BackgroundWorkers/Volo/Abp/BackgroundWorkers/AsyncPeriodicBackgroundWorkerBase.cs

@@ -12,6 +12,7 @@ public abstract class AsyncPeriodicBackgroundWorkerBase : BackgroundWorkerBase
 {
     protected IServiceScopeFactory ServiceScopeFactory { get; }
     protected AbpAsyncTimer Timer { get; }
+    protected CancellationToken StartCancellationToken { get; set; }
 
     protected AsyncPeriodicBackgroundWorkerBase(
         AbpAsyncTimer timer,
@@ -22,13 +23,15 @@ public abstract class AsyncPeriodicBackgroundWorkerBase : BackgroundWorkerBase
         Timer.Elapsed = Timer_Elapsed;
     }
 
-    public override async Task StartAsync(CancellationToken cancellationToken = default)
+    public async override Task StartAsync(CancellationToken cancellationToken = default)
     {
+        StartCancellationToken = cancellationToken;
+
         await base.StartAsync(cancellationToken);
         Timer.Start(cancellationToken);
     }
 
-    public override async Task StopAsync(CancellationToken cancellationToken = default)
+    public async override Task StopAsync(CancellationToken cancellationToken = default)
     {
         Timer.Stop(cancellationToken);
         await base.StopAsync(cancellationToken);
@@ -36,16 +39,16 @@ public abstract class AsyncPeriodicBackgroundWorkerBase : BackgroundWorkerBase
 
     private async Task Timer_Elapsed(AbpAsyncTimer timer)
     {
-        await DoWorkAsync();
+        await DoWorkAsync(StartCancellationToken);
     }
 
-    private async Task DoWorkAsync()
+    private async Task DoWorkAsync(CancellationToken cancellationToken = default)
     {
         using (var scope = ServiceScopeFactory.CreateScope())
         {
             try
             {
-                await DoWorkAsync(new PeriodicBackgroundWorkerContext(scope.ServiceProvider));
+                await DoWorkAsync(new PeriodicBackgroundWorkerContext(scope.ServiceProvider, cancellationToken));
             }
             catch (Exception ex)
             {

framework/src/Volo.Abp.BackgroundWorkers/Volo/Abp/BackgroundWorkers/BackgroundWorkerManager.cs

@@ -26,13 +26,13 @@ public class BackgroundWorkerManager : IBackgroundWorkerManager, ISingletonDepen
         _backgroundWorkers = new List<IBackgroundWorker>();
     }
 
-    public virtual async Task AddAsync(IBackgroundWorker worker)
+    public virtual async Task AddAsync(IBackgroundWorker worker, CancellationToken cancellationToken = default)
     {
         _backgroundWorkers.Add(worker);
 
         if (IsRunning)
         {
-            await worker.StartAsync();
+            await worker.StartAsync(cancellationToken);
         }
     }
 

framework/src/Volo.Abp.BackgroundWorkers/Volo/Abp/BackgroundWorkers/BackgroundWorkersApplicationInitializationContextExtensions.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using JetBrains.Annotations;
 using Microsoft.Extensions.DependencyInjection;
@@ -7,17 +8,17 @@ namespace Volo.Abp.BackgroundWorkers;
 
 public static class BackgroundWorkersApplicationInitializationContextExtensions
 {
-    public async static Task<ApplicationInitializationContext> AddBackgroundWorkerAsync<TWorker>([NotNull] this ApplicationInitializationContext context)
+    public async static Task<ApplicationInitializationContext> AddBackgroundWorkerAsync<TWorker>([NotNull] this ApplicationInitializationContext context, CancellationToken cancellationToken = default)
         where TWorker : IBackgroundWorker
     {
         Check.NotNull(context, nameof(context));
 
-        await context.AddBackgroundWorkerAsync(typeof(TWorker));
+        await context.AddBackgroundWorkerAsync(typeof(TWorker), cancellationToken: cancellationToken);
 
         return context;
     }
 
-    public async static Task<ApplicationInitializationContext> AddBackgroundWorkerAsync([NotNull] this ApplicationInitializationContext context, [NotNull] Type workerType)
+    public async static Task<ApplicationInitializationContext> AddBackgroundWorkerAsync([NotNull] this ApplicationInitializationContext context, [NotNull] Type workerType, CancellationToken cancellationToken = default)
     {
         Check.NotNull(context, nameof(context));
         Check.NotNull(workerType, nameof(workerType));
@@ -29,9 +30,7 @@ public static class BackgroundWorkersApplicationInitializationContextExtensions
 
         await context.ServiceProvider
             .GetRequiredService<IBackgroundWorkerManager>()
-            .AddAsync(
-                (IBackgroundWorker)context.ServiceProvider.GetRequiredService(workerType)
-            );
+            .AddAsync((IBackgroundWorker)context.ServiceProvider.GetRequiredService(workerType), cancellationToken);
 
         return context;
     }

framework/src/Volo.Abp.BackgroundWorkers/Volo/Abp/BackgroundWorkers/IBackgroundWorkerManager.cs

@@ -1,4 +1,5 @@
-using System.Threading.Tasks;
+using System.Threading;
+using System.Threading.Tasks;
 using Volo.Abp.Threading;
 
 namespace Volo.Abp.BackgroundWorkers;
@@ -14,5 +15,6 @@ public interface IBackgroundWorkerManager : IRunnable
     /// <param name="worker">
     /// The worker. It should be resolved from IOC.
     /// </param>
-    Task AddAsync(IBackgroundWorker worker);
+    /// <param name="cancellationToken"></param>
+    Task AddAsync(IBackgroundWorker worker, CancellationToken cancellationToken = default);
 }

framework/src/Volo.Abp.BackgroundWorkers/Volo/Abp/BackgroundWorkers/PeriodicBackgroundWorkerContext.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 
 namespace Volo.Abp.BackgroundWorkers;
 
@@ -6,8 +7,17 @@ public class PeriodicBackgroundWorkerContext
 {
     public IServiceProvider ServiceProvider { get; }
 
+    public CancellationToken CancellationToken { get; }
+
     public PeriodicBackgroundWorkerContext(IServiceProvider serviceProvider)
     {
         ServiceProvider = serviceProvider;
+        CancellationToken = default;
+    }
+
+    public PeriodicBackgroundWorkerContext(IServiceProvider serviceProvider, CancellationToken cancellationToken)
+    {
+        ServiceProvider = serviceProvider;
+        CancellationToken = cancellationToken;
     }
 }