Browse Source

Moved background manager from the old solution.

pull/395/head
Halil ibrahim Kalkan 8 years ago
parent
commit
3e4798a7e2
  1. 1
      framework/src/Volo.Abp.Auditing/Volo/Abp/Auditing/JsonNetAuditSerializer.cs
  2. 4
      framework/src/Volo.Abp.BackgroundJobs/Volo.Abp.BackgroundJobs.csproj
  3. 12
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobsModule.cs
  4. 19
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJob.cs
  5. 44
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobException.cs
  6. 126
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobInfo.cs
  7. 182
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobManager.cs
  8. 26
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobManagerExtensions.cs
  9. 12
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobOptions.cs
  10. 33
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobPriority.cs
  11. 14
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJob.cs
  12. 32
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobManager.cs
  13. 11
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobSerializer.cs
  14. 46
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobStore.cs
  15. 30
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/JsonBackgroundJobSerializer.cs
  16. 39
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/NullBackgroundJobStore.cs

1
framework/src/Volo.Abp.Auditing/Volo/Abp/Auditing/JsonNetAuditSerializer.cs

@ -4,6 +4,7 @@ using Volo.Abp.DependencyInjection;
namespace Volo.Abp.Auditing namespace Volo.Abp.Auditing
{ {
//TODO: Rename to JsonAuditSerializer
public class JsonNetAuditSerializer : IAuditSerializer, ITransientDependency public class JsonNetAuditSerializer : IAuditSerializer, ITransientDependency
{ {
protected AbpAuditingOptions Options; protected AbpAuditingOptions Options;

4
framework/src/Volo.Abp.BackgroundJobs/Volo.Abp.BackgroundJobs.csproj

@ -14,7 +14,11 @@
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<ProjectReference Include="..\Volo.Abp.BackgroundWorkers\Volo.Abp.BackgroundWorkers.csproj" />
<ProjectReference Include="..\Volo.Abp.Core\Volo.Abp.Core.csproj" /> <ProjectReference Include="..\Volo.Abp.Core\Volo.Abp.Core.csproj" />
<ProjectReference Include="..\Volo.Abp.Guids\Volo.Abp.Guids.csproj" />
<ProjectReference Include="..\Volo.Abp.Json\Volo.Abp.Json.csproj" />
<ProjectReference Include="..\Volo.Abp.Timing\Volo.Abp.Timing.csproj" />
</ItemGroup> </ItemGroup>
</Project> </Project>

12
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobsModule.cs

@ -1,8 +1,18 @@
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Volo.Abp.BackgroundWorkers;
using Volo.Abp.Guids;
using Volo.Abp.Json;
using Volo.Abp.Modularity; using Volo.Abp.Modularity;
using Volo.Abp.Timing;
namespace Volo.Abp.BackgroundJobs namespace Volo.Abp.BackgroundJobs
{ {
[DependsOn(
typeof(AbpBackgroundWorkersModule),
typeof(AbpTimingModule),
typeof(AbpJsonModule),
typeof(AbpGuidsModule)
)]
public class AbpBackgroundJobsModule : AbpModule public class AbpBackgroundJobsModule : AbpModule
{ {
public override void ConfigureServices(ServiceConfigurationContext context) public override void ConfigureServices(ServiceConfigurationContext context)
@ -10,4 +20,4 @@ namespace Volo.Abp.BackgroundJobs
context.Services.AddAssemblyOf<AbpBackgroundJobsModule>(); context.Services.AddAssemblyOf<AbpBackgroundJobsModule>();
} }
} }
} }

19
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJob.cs

@ -0,0 +1,19 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
namespace Volo.Abp.BackgroundJobs
{
public abstract class BackgroundJob<TArgs> : IBackgroundJob<TArgs>
{
//TODO: Add UOW, Localization and other useful properties..?
public ILogger<BackgroundJob<TArgs>> Logger { get; set; }
protected BackgroundJob()
{
Logger = NullLogger<BackgroundJob<TArgs>>.Instance;
}
public abstract void Execute(TArgs args);
}
}

44
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobException.cs

@ -0,0 +1,44 @@
using System;
using System.Runtime.Serialization;
using JetBrains.Annotations;
namespace Volo.Abp.BackgroundJobs
{
[Serializable]
public class BackgroundJobException : AbpException
{
[CanBeNull]
public BackgroundJobInfo BackgroundJob { get; set; }
[CanBeNull]
public object JobObject { get; set; }
/// <summary>
/// Creates a new <see cref="BackgroundJobException"/> object.
/// </summary>
public BackgroundJobException()
{
}
/// <summary>
/// Creates a new <see cref="BackgroundJobException"/> object.
/// </summary>
public BackgroundJobException(SerializationInfo serializationInfo, StreamingContext context)
: base(serializationInfo, context)
{
}
/// <summary>
/// Creates a new <see cref="BackgroundJobException"/> object.
/// </summary>
/// <param name="message">Exception message</param>
/// <param name="innerException">Inner exception</param>
public BackgroundJobException(string message, Exception innerException)
: base(message, innerException)
{
}
}
}

126
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobInfo.cs

@ -0,0 +1,126 @@
using System;
using System.ComponentModel.DataAnnotations;
using Volo.Abp.Timing;
namespace Volo.Abp.BackgroundJobs
{
/// <summary>
/// Represents a background job info that is used to persist jobs.
/// </summary>
public class BackgroundJobInfo
{
/// <summary>
/// Maximum length of <see cref="JobType"/>.
/// Value: 512.
/// </summary>
public const int MaxJobTypeLength = 512;
/// <summary>
/// Maximum length of <see cref="JobArgs"/>.
/// Value: 1 MB (1,048,576 bytes).
/// </summary>
public const int MaxJobArgsLength = 1024 * 1024;
/// <summary>
/// Default duration (as seconds) for the first wait on a failure.
/// Default value: 60 (1 minutes).
/// </summary>
public static int DefaultFirstWaitDuration { get; set; } //TODO: Move to configuration
/// <summary>
/// Default timeout value (as seconds) for a job before it's abandoned (<see cref="IsAbandoned"/>).
/// Default value: 172,800 (2 days).
/// </summary>
public static int DefaultTimeout { get; set; } //TODO: Move to configuration
/// <summary>
/// Default wait factor for execution failures.
/// This amount is multiplated by last wait time to calculate next wait time.
/// Default value: 2.0.
/// </summary>
public static double DefaultWaitFactor { get; set; } //TODO: Move to configuration
public Guid Id { get; set; }
/// <summary>
/// Type of the job.
/// It's AssemblyQualifiedName of job type.
/// </summary>
[Required]
[StringLength(MaxJobTypeLength)]
public virtual string JobType { get; set; }
/// <summary>
/// Job arguments as JSON string.
/// </summary>
[Required]
[MaxLength(MaxJobArgsLength)]
public virtual string JobArgs { get; set; } //TODO: Consider to conver to byte[]
/// <summary>
/// Try count of this job.
/// A job is re-tried if it fails.
/// </summary>
public virtual short TryCount { get; set; }
/// <summary>
/// Creation time of this job.
/// </summary>
public virtual DateTime CreationTime { get; set; }
/// <summary>
/// Next try time of this job.
/// </summary>
public virtual DateTime NextTryTime { get; set; }
/// <summary>
/// Last try time of this job.
/// </summary>
public virtual DateTime? LastTryTime { get; set; }
/// <summary>
/// This is true if this job is continously failed and will not be executed again.
/// </summary>
public virtual bool IsAbandoned { get; set; }
/// <summary>
/// Priority of this job.
/// </summary>
public virtual BackgroundJobPriority Priority { get; set; }
static BackgroundJobInfo()
{
DefaultFirstWaitDuration = 60;
DefaultTimeout = 172800;
DefaultWaitFactor = 2.0;
}
/// <summary>
/// Initializes a new instance of the <see cref="BackgroundJobInfo"/> class.
/// </summary>
public BackgroundJobInfo()
{
Priority = BackgroundJobPriority.Normal;
}
/// <summary>
/// Calculates next try time if a job fails.
/// Returns null if it will not wait anymore and job should be abandoned.
/// </summary>
/// <returns></returns>
public virtual DateTime? CalculateNextTryTime(IClock clock) //TODO: Move to another place to override easier
{
var nextWaitDuration = DefaultFirstWaitDuration * (Math.Pow(DefaultWaitFactor, TryCount - 1));
var nextTryDate = LastTryTime.HasValue
? LastTryTime.Value.AddSeconds(nextWaitDuration)
: clock.Now.AddSeconds(nextWaitDuration);
if (nextTryDate.Subtract(CreationTime).TotalSeconds > DefaultTimeout)
{
return null;
}
return nextTryDate;
}
}
}

182
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobManager.cs

@ -0,0 +1,182 @@
using System;
using System.Diagnostics;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Volo.Abp.BackgroundWorkers;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Guids;
using Volo.Abp.Threading;
using Volo.Abp.Timing;
namespace Volo.Abp.BackgroundJobs
{
/// <summary>
/// Default implementation of <see cref="IBackgroundJobManager"/>.
/// </summary>
public class BackgroundJobManager : PeriodicBackgroundWorkerBase, IBackgroundJobManager, ISingletonDependency
{
/// <summary>
/// Interval between polling jobs from <see cref="IBackgroundJobStore"/>.
/// Default value: 5000 (5 seconds).
/// </summary>
public static int JobPollPeriod { get; set; } //TODO: Move to options
protected IServiceProvider ServiceProvider { get; }
protected IClock Clock { get; }
protected IBackgroundJobSerializer Serializer { get; }
protected IGuidGenerator GuidGenerator { get; }
protected IBackgroundJobStore Store { get; }
static BackgroundJobManager()
{
JobPollPeriod = 5000;
}
/// <summary>
/// Initializes a new instance of the <see cref="BackgroundJobManager"/> class.
/// </summary>
public BackgroundJobManager(
IServiceProvider serviceProvider,
IClock clock,
IBackgroundJobSerializer serializer,
IBackgroundJobStore store,
IGuidGenerator guidGenerator,
AbpTimer timer)
: base(timer)
{
ServiceProvider = serviceProvider;
Clock = clock;
Serializer = serializer;
GuidGenerator = guidGenerator;
Store = store;
Timer.Period = JobPollPeriod;
}
public async Task<Guid> EnqueueAsync<TJob, TArgs>(TArgs args, BackgroundJobPriority priority = BackgroundJobPriority.Normal, TimeSpan? delay = null)
where TJob : IBackgroundJob<TArgs>
{
var jobInfo = new BackgroundJobInfo
{
Id = GuidGenerator.Create(),
JobType = typeof(TJob).AssemblyQualifiedName,
JobArgs = Serializer.Serialize(args),
Priority = priority,
CreationTime = Clock.Now,
NextTryTime = Clock.Now
};
if (delay.HasValue)
{
jobInfo.NextTryTime = Clock.Now.Add(delay.Value);
}
await Store.InsertAsync(jobInfo);
return jobInfo.Id;
}
public async Task<bool> DeleteAsync(Guid jobId)
{
var jobInfo = await Store.GetAsync(jobId);
if (jobInfo == null)
{
return false;
}
await Store.DeleteAsync(jobInfo);
return true;
}
protected override void DoWork()
{
var waitingJobs = AsyncHelper.RunSync(() => Store.GetWaitingJobsAsync(1000));
foreach (var job in waitingJobs)
{
TryProcessJob(job);
}
}
private void TryProcessJob(BackgroundJobInfo jobInfo)
{
try
{
jobInfo.TryCount++;
jobInfo.LastTryTime = Clock.Now;
var jobType = Type.GetType(jobInfo.JobType);
using (var scope = ServiceProvider.CreateScope())
{
var job = scope.ServiceProvider.GetService(jobType);
if (job == null)
{
throw new AbpException("JobType is not registered: " + jobType);
}
//TODO: Type check for the job object
var jobExecuteMethod = job.GetType().GetMethod("Execute");
Debug.Assert(jobExecuteMethod != null, nameof(jobExecuteMethod) + " != null");
var argsType = jobExecuteMethod.GetParameters()[0].ParameterType;
var argsObj = Serializer.Deserialize(jobInfo.JobArgs, argsType);
try
{
jobExecuteMethod.Invoke(job, new[] { argsObj });
AsyncHelper.RunSync(() => Store.DeleteAsync(jobInfo));
}
catch (Exception ex)
{
Logger.LogException(ex);
var nextTryTime = jobInfo.CalculateNextTryTime(Clock);
if (nextTryTime.HasValue)
{
jobInfo.NextTryTime = nextTryTime.Value;
}
else
{
jobInfo.IsAbandoned = true;
}
TryUpdate(jobInfo);
var backgroundJobException = new BackgroundJobException(
"A background job execution is failed. See inner exception for details. See BackgroundJob property to get information on the background job.",
ex
)
{
BackgroundJob = jobInfo,
JobObject = job
};
//TODO: Somehow trigger an event for the exception (may create an Volo.Abp.ExceptionHandling package)!
}
}
}
catch (Exception ex)
{
Logger.LogException(ex);
jobInfo.IsAbandoned = true;
TryUpdate(jobInfo);
}
}
private void TryUpdate(BackgroundJobInfo jobInfo)
{
try
{
Store.UpdateAsync(jobInfo);
}
catch (Exception updateEx)
{
Logger.LogException(updateEx);
}
}
}
}

26
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobManagerExtensions.cs

@ -0,0 +1,26 @@
using System;
using Volo.Abp.Threading;
namespace Volo.Abp.BackgroundJobs
{
/// <summary>
/// Some extension methods for <see cref="IBackgroundJobManager"/>.
/// </summary>
public static class BackgroundJobManagerExtensions
{
/// <summary>
/// Enqueues a job to be executed.
/// </summary>
/// <typeparam name="TJob">Type of the job.</typeparam>
/// <typeparam name="TArgs">Type of the arguments of job.</typeparam>
/// <param name="backgroundJobManager">Background job manager reference</param>
/// <param name="args">Job arguments.</param>
/// <param name="priority">Job priority.</param>
/// <param name="delay">Job delay (wait duration before first try).</param>
public static void Enqueue<TJob, TArgs>(this IBackgroundJobManager backgroundJobManager, TArgs args, BackgroundJobPriority priority = BackgroundJobPriority.Normal, TimeSpan? delay = null)
where TJob : IBackgroundJob<TArgs>
{
AsyncHelper.RunSync(() => backgroundJobManager.EnqueueAsync<TJob, TArgs>(args, priority, delay));
}
}
}

12
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobOptions.cs

@ -0,0 +1,12 @@
namespace Volo.Abp.BackgroundJobs
{
public class BackgroundJobOptions
{
public bool IsJobExecutionEnabled { get; set; }
public BackgroundJobOptions()
{
IsJobExecutionEnabled = true;
}
}
}

33
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobPriority.cs

@ -0,0 +1,33 @@
namespace Volo.Abp.BackgroundJobs
{
/// <summary>
/// Priority of a background job.
/// </summary>
public enum BackgroundJobPriority : byte
{
/// <summary>
/// Low.
/// </summary>
Low = 5,
/// <summary>
/// Below normal.
/// </summary>
BelowNormal = 10,
/// <summary>
/// Normal (default).
/// </summary>
Normal = 15,
/// <summary>
/// Above normal.
/// </summary>
AboveNormal = 20,
/// <summary>
/// High.
/// </summary>
High = 25
}
}

14
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJob.cs

@ -0,0 +1,14 @@
namespace Volo.Abp.BackgroundJobs
{
/// <summary>
/// Defines interface of a background job.
/// </summary>
public interface IBackgroundJob<in TArgs>
{
/// <summary>
/// Executes the job with the <see cref="args"/>.
/// </summary>
/// <param name="args">Job arguments.</param>
void Execute(TArgs args);
}
}

32
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobManager.cs

@ -0,0 +1,32 @@
using System;
using System.Threading.Tasks;
using Volo.Abp.BackgroundWorkers;
namespace Volo.Abp.BackgroundJobs
{
//TODO: Create a non-generic EnqueueAsync extension method to IBackgroundJobManager which takes types as input parameters rather than generic parameters.
/// <summary>
/// Defines interface of a job manager.
/// </summary>
public interface IBackgroundJobManager : IBackgroundWorker
{
/// <summary>
/// Enqueues a job to be executed.
/// </summary>
/// <typeparam name="TJob">Type of the job.</typeparam>
/// <typeparam name="TArgs">Type of the arguments of job.</typeparam>
/// <param name="args">Job arguments.</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<Guid> EnqueueAsync<TJob, TArgs>(TArgs args, BackgroundJobPriority priority = BackgroundJobPriority.Normal, TimeSpan? delay = null)
where TJob : IBackgroundJob<TArgs>;
/// <summary>
/// Deletes a job with the specified jobId.
/// </summary>
/// <param name="jobId">The Job Unique Identifier.</param>
/// <returns><c>True</c> on a successfull state transition, <c>false</c> otherwise.</returns>
Task<bool> DeleteAsync(Guid jobId);
}
}

11
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobSerializer.cs

@ -0,0 +1,11 @@
using System;
namespace Volo.Abp.BackgroundJobs
{
public interface IBackgroundJobSerializer
{
string Serialize(object obj);
object Deserialize(string value, Type type);
}
}

46
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobStore.cs

@ -0,0 +1,46 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Volo.Abp.BackgroundJobs
{
/// <summary>
/// Defines interface to store/get background jobs.
/// </summary>
public interface IBackgroundJobStore
{
/// <summary>
/// Gets a BackgroundJobInfo based on the given jobId.
/// </summary>
/// <param name="jobId">The Job Unique Identifier.</param>
/// <returns>The BackgroundJobInfo object.</returns>
Task<BackgroundJobInfo> GetAsync(Guid jobId);
/// <summary>
/// Inserts a background job.
/// </summary>
/// <param name="jobInfo">Job information.</param>
Task InsertAsync(BackgroundJobInfo jobInfo);
/// <summary>
/// Gets waiting jobs. It should get jobs based on these:
/// Conditions: !IsAbandoned And NextTryTime &lt;= Clock.Now.
/// Order by: Priority DESC, TryCount ASC, NextTryTime ASC.
/// Maximum result: <paramref name="maxResultCount"/>.
/// </summary>
/// <param name="maxResultCount">Maximum result count.</param>
Task<List<BackgroundJobInfo>> GetWaitingJobsAsync(int maxResultCount);
/// <summary>
/// Deletes a job.
/// </summary>
/// <param name="jobInfo">Job information.</param>
Task DeleteAsync(BackgroundJobInfo jobInfo);
/// <summary>
/// Updates a job.
/// </summary>
/// <param name="jobInfo">Job information.</param>
Task UpdateAsync(BackgroundJobInfo jobInfo);
}
}

30
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/JsonBackgroundJobSerializer.cs

@ -0,0 +1,30 @@
using System;
using Volo.Abp.Json;
namespace Volo.Abp.BackgroundJobs
{
public class JsonBackgroundJobSerializer : IBackgroundJobSerializer
{
private readonly IJsonSerializer _jsonSerializer;
public JsonBackgroundJobSerializer(IJsonSerializer jsonSerializer)
{
_jsonSerializer = jsonSerializer;
}
public string Serialize(object obj)
{
return _jsonSerializer.Serialize(obj);
}
public object Deserialize(string value, Type type)
{
return _jsonSerializer.Deserialize(type, value);
}
public T Deserialize<T>(string value)
{
return _jsonSerializer.Deserialize<T>(value);
}
}
}

39
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/NullBackgroundJobStore.cs

@ -0,0 +1,39 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Volo.Abp.BackgroundJobs
{
/// <summary>
/// Null pattern implementation of <see cref="IBackgroundJobStore"/>.
/// It's used if <see cref="IBackgroundJobStore"/> is not implemented by actual persistent store
/// and job execution is not enabled (<see cref="IBackgroundJobOptions.IsJobExecutionEnabled"/>) for the application.
/// </summary>
public class NullBackgroundJobStore : IBackgroundJobStore
{
public Task<BackgroundJobInfo> GetAsync(Guid jobId)
{
return Task.FromResult((BackgroundJobInfo)null);
}
public Task InsertAsync(BackgroundJobInfo jobInfo)
{
return Task.CompletedTask;
}
public Task<List<BackgroundJobInfo>> GetWaitingJobsAsync(int maxResultCount)
{
return Task.FromResult(new List<BackgroundJobInfo>());
}
public Task DeleteAsync(BackgroundJobInfo jobInfo)
{
return Task.CompletedTask;
}
public Task UpdateAsync(BackgroundJobInfo jobInfo)
{
return Task.CompletedTask;
}
}
}
Loading…
Cancel
Save