diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs index c01f933781..18862c8b1d 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs @@ -48,6 +48,12 @@ public class BackgroundJobRecord : AggregateRoot, IHasCreationTime /// public virtual bool IsAbandoned { get; set; } + /// + /// The time this job was completed successfully. When set, the job is kept as history and excluded + /// from the waiting jobs query (set only when successful job persistence is enabled). + /// + public virtual DateTime? CompletionTime { get; set; } + /// /// Priority of this job. /// diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs index c7dbd6fbef..9d24c6c236 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; using Volo.Abp.DependencyInjection; using Volo.Abp.ObjectMapping; @@ -22,9 +23,13 @@ public class BackgroundJobStore : IBackgroundJobStore, ITransientDependency public virtual async Task FindAsync(Guid jobId) { - return ObjectMapper.Map( - await BackgroundJobRepository.FindAsync(jobId) - ); + var backgroundJobRecord = await BackgroundJobRepository.FindAsync(jobId); + if (backgroundJobRecord == null) + { + return null!; + } + + return ObjectMapper.Map(backgroundJobRecord); } public virtual async Task InsertAsync(BackgroundJobInfo jobInfo) @@ -34,18 +39,37 @@ public class BackgroundJobStore : IBackgroundJobStore, ITransientDependency ); } - public virtual async Task> GetWaitingJobsAsync(string applicationName, int maxResultCount) + public virtual async Task> GetWaitingJobsAsync(string? applicationName, int maxResultCount) { return ObjectMapper.Map, List>( await BackgroundJobRepository.GetWaitingListAsync(applicationName, maxResultCount) ); } + public virtual async Task> GetWaitingJobsAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter) + { + return ObjectMapper.Map, List>( + await BackgroundJobRepository.GetWaitingListAsync(applicationName, maxResultCount, jobNameFilter) + ); + } + public virtual async Task DeleteAsync(Guid jobId) { await BackgroundJobRepository.DeleteAsync(jobId); } + public virtual async Task DeleteAsync( + string? applicationName, + DateTime completedBefore, + int maxResultCount, + CancellationToken cancellationToken = default) + { + return await BackgroundJobRepository.DeleteAsync(applicationName, completedBefore, maxResultCount, cancellationToken); + } + public virtual async Task UpdateAsync(BackgroundJobInfo jobInfo) { var backgroundJobRecord = await BackgroundJobRepository.FindAsync(jobInfo.Id); diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs index 9ac5cef9bf..a605582201 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs @@ -5,7 +5,7 @@ using System.Threading.Tasks; using Riok.Mapperly.Abstractions; using Volo.Abp.Mapperly; -namespace Volo.Abp.BackgroundJobs; +namespace Volo.Abp.BackgroundJobs; [Mapper(RequiredMappingStrategy = RequiredMappingStrategy.Target)] public partial class BackgroundJobInfoToBackgroundJobRecordMapper @@ -31,6 +31,6 @@ public partial class BackgroundJobRecordToBackgroundJobInfoMapper : MapperBase { public override partial BackgroundJobInfo Map(BackgroundJobRecord source); - + public override partial void Map(BackgroundJobRecord source, BackgroundJobInfo destination); } diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs index 7e189df5c8..d4e8df2d83 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs @@ -2,12 +2,26 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; -using JetBrains.Annotations; using Volo.Abp.Domain.Repositories; namespace Volo.Abp.BackgroundJobs; public interface IBackgroundJobRepository : IBasicRepository { - Task> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default); + Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + CancellationToken cancellationToken = default); + + Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter, + CancellationToken cancellationToken = default); + + Task DeleteAsync( + string? applicationName, + DateTime completedBefore, + int maxResultCount, + CancellationToken cancellationToken = default); } diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs index 53a2c53843..70a73a8562 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs @@ -29,9 +29,10 @@ public static class BackgroundJobsDbContextModelCreatingExtensions b.Property(x => x.NextTryTime); b.Property(x => x.LastTryTime); b.Property(x => x.IsAbandoned).HasDefaultValue(false); + b.Property(x => x.CompletionTime); b.Property(x => x.Priority).HasDefaultValue(BackgroundJobPriority.Normal).HasSentinel(BackgroundJobPriority.Normal); - b.HasIndex(x => new { x.IsAbandoned, x.NextTryTime }); + b.HasIndex(x => new { x.ApplicationName, x.CompletionTime, x.IsAbandoned, x.NextTryTime }); b.ApplyObjectExtensionMappings(); }); diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs index a81de3cdfb..5b55a7adf8 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs @@ -3,7 +3,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; -using JetBrains.Annotations; using Microsoft.EntityFrameworkCore; using Volo.Abp.Domain.Repositories.EntityFrameworkCore; using Volo.Abp.EntityFrameworkCore; @@ -23,20 +22,71 @@ public class EfCoreBackgroundJobRepository : EfCoreRepository> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default) + public virtual Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + CancellationToken cancellationToken = default) { - return await (await GetWaitingListQueryAsync(applicationName, maxResultCount)).ToListAsync(GetCancellationToken(cancellationToken)); + return GetWaitingListAsync(applicationName, maxResultCount, null, cancellationToken); } - protected virtual async Task> GetWaitingListQueryAsync([CanBeNull] string applicationName, int maxResultCount) + public virtual async Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter, + CancellationToken cancellationToken = default) + { + return await (await GetWaitingListQueryAsync(applicationName, maxResultCount, jobNameFilter)) + .ToListAsync(GetCancellationToken(cancellationToken)); + } + + protected virtual async Task> GetWaitingListQueryAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter = null) { var now = Clock.Now; - return (await GetDbSetAsync()) + var filter = jobNameFilter ?? BackgroundJobNameFilter.None; + var jobNames = filter.JobNames.ToList(); + + var query = (await GetDbSetAsync()) .Where(t => t.ApplicationName == applicationName) - .Where(t => !t.IsAbandoned && t.NextTryTime <= now) + .Where(t => !t.IsAbandoned && t.CompletionTime == null && t.NextTryTime <= now); + + if (filter.Mode == BackgroundJobNameFilterMode.Include) + { + query = query.Where(t => jobNames.Contains(t.JobName)); + } + else if (filter.Mode == BackgroundJobNameFilterMode.Exclude) + { + query = query.Where(t => !jobNames.Contains(t.JobName)); + } + + return query .OrderByDescending(t => t.Priority) .ThenBy(t => t.TryCount) .ThenBy(t => t.NextTryTime) .Take(maxResultCount); } + + public virtual async Task DeleteAsync(string? applicationName, DateTime completedBefore, int maxResultCount, CancellationToken cancellationToken = default) + { + var token = GetCancellationToken(cancellationToken); + var dbSet = await GetDbSetAsync(); + + var ids = await dbSet + .Where(t => t.ApplicationName == applicationName) + .Where(t => t.CompletionTime != null && t.CompletionTime < completedBefore) + .OrderBy(t => t.CompletionTime) + .Select(t => t.Id) + .Take(maxResultCount) + .ToListAsync(token); + + if (ids.Count == 0) + { + return 0; + } + + return await dbSet.Where(t => ids.Contains(t.Id)).ExecuteDeleteAsync(token); + } } diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs index 71a5680524..af4ab7ef89 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs @@ -3,7 +3,6 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using System.Linq; -using JetBrains.Annotations; using MongoDB.Driver; using MongoDB.Driver.Linq; using Volo.Abp.Domain.Repositories.MongoDB; @@ -24,20 +23,78 @@ public class MongoBackgroundJobRepository : MongoDbRepository> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default) + public virtual Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + CancellationToken cancellationToken = default) { - return await (await GetWaitingListQuery(applicationName, maxResultCount, cancellationToken)).ToListAsync(GetCancellationToken(cancellationToken)); + return GetWaitingListAsync(applicationName, maxResultCount, null, cancellationToken); } - protected virtual async Task> GetWaitingListQuery([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default) + public virtual async Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter, + CancellationToken cancellationToken = default) + { + return await (await GetWaitingListQuery(applicationName, maxResultCount, jobNameFilter, cancellationToken)) + .ToListAsync(GetCancellationToken(cancellationToken)); + } + + protected virtual async Task> GetWaitingListQuery( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter = null, + CancellationToken cancellationToken = default) { var now = Clock.Now; - return (await GetQueryableAsync(cancellationToken)) + var filter = jobNameFilter ?? BackgroundJobNameFilter.None; + var jobNames = filter.JobNames.ToList(); + + var query = (await GetQueryableAsync(cancellationToken)) .Where(t => t.ApplicationName == applicationName) - .Where(t => !t.IsAbandoned && t.NextTryTime <= now) + .Where(t => !t.IsAbandoned && t.CompletionTime == null && t.NextTryTime <= now); + + if (filter.Mode == BackgroundJobNameFilterMode.Include) + { + query = query.Where(t => jobNames.Contains(t.JobName)); + } + else if (filter.Mode == BackgroundJobNameFilterMode.Exclude) + { + query = query.Where(t => !jobNames.Contains(t.JobName)); + } + + return query .OrderByDescending(t => t.Priority) .ThenBy(t => t.TryCount) .ThenBy(t => t.NextTryTime) .Take(maxResultCount); } + + public virtual async Task DeleteAsync(string? applicationName, DateTime completedBefore, int maxResultCount, CancellationToken cancellationToken = default) + { + var token = GetCancellationToken(cancellationToken); + + var ids = await (await GetQueryableAsync(token)) + .Where(t => t.ApplicationName == applicationName) + .Where(t => t.CompletionTime != null && t.CompletionTime < completedBefore) + .OrderBy(t => t.CompletionTime) + .Select(t => t.Id) + .Take(maxResultCount) + .ToListAsync(token); + + if (ids.Count == 0) + { + return 0; + } + + var dbContext = await GetDbContextAsync(token); + var collection = dbContext.Collection(); + + var result = dbContext.SessionHandle != null + ? await collection.DeleteManyAsync(dbContext.SessionHandle, x => ids.Contains(x.Id), cancellationToken: token) + : await collection.DeleteManyAsync(x => ids.Contains(x.Id), cancellationToken: token); + + return (int)result.DeletedCount; + } }