Browse Source

Merge pull request #6417 from abpframework/maliming/6385

Discard the task result in Timer_Elapsed.
pull/6532/head
Halil İbrahim Kalkan 6 years ago
committed by GitHub
parent
commit
f107600e27
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
  1. 4
      framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorker.cs
  2. 2
      framework/src/Volo.Abp.BackgroundWorkers.Quartz/Volo/Abp/BackgroundWorkers/Quartz/QuartzBackgroundWorkerAdapter.cs
  3. 25
      framework/src/Volo.Abp.BackgroundWorkers/Volo/Abp/BackgroundWorkers/AsyncPeriodicBackgroundWorkerBase.cs
  4. 6
      framework/src/Volo.Abp.BackgroundWorkers/Volo/Abp/BackgroundWorkers/PeriodicBackgroundWorkerBase.cs
  5. 2
      framework/src/Volo.Abp.Caching/Volo/Abp/Caching/DistributedCache.cs
  6. 12
      framework/src/Volo.Abp.Cli/Volo/Abp/Cli/Program.cs
  7. 51
      framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs
  8. 127
      framework/src/Volo.Abp.Threading/Volo/Abp/Threading/AbpAsyncTimer.cs
  9. 2
      framework/src/Volo.Abp.Threading/Volo/Abp/Threading/AbpTimer.cs
  10. 4
      framework/test/Volo.Abp.BlobStoring.Aws.Tests/Volo/Abp/BlobStoring/Aws/AbpBlobStoringAwsTestModule.cs
  11. 7
      modules/identityserver/src/Volo.Abp.IdentityServer.Domain/Volo/Abp/IdentityServer/Tokens/TokenCleanupBackgroundWorker.cs

4
framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorker.cs

@ -17,7 +17,7 @@ namespace Volo.Abp.BackgroundJobs
protected AbpBackgroundJobWorkerOptions WorkerOptions { get; }
public BackgroundJobWorker(
AbpTimer timer,
AbpAsyncTimer timer,
IOptions<AbpBackgroundJobOptions> jobOptions,
IOptions<AbpBackgroundJobWorkerOptions> workerOptions,
IServiceScopeFactory serviceScopeFactory)
@ -113,4 +113,4 @@ namespace Volo.Abp.BackgroundJobs
return nextTryDate;
}
}
}
}

2
framework/src/Volo.Abp.BackgroundWorkers.Quartz/Volo/Abp/BackgroundWorkers/Quartz/QuartzBackgroundWorkerAdapter.cs

@ -34,7 +34,7 @@ namespace Volo.Abp.BackgroundWorkers.Quartz
throw new ArgumentException($"{nameof(worker)} type is different from the generic type");
}
var timer = (AbpTimer) worker.GetType().GetProperty("Timer", BindingFlags.Instance | BindingFlags.NonPublic)?.GetValue(worker);
var timer = (AbpAsyncTimer) worker.GetType().GetProperty("Timer", BindingFlags.Instance | BindingFlags.NonPublic)?.GetValue(worker);
period = timer?.Period;
}
else

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

@ -11,15 +11,15 @@ namespace Volo.Abp.BackgroundWorkers
public abstract class AsyncPeriodicBackgroundWorkerBase : BackgroundWorkerBase
{
protected IServiceScopeFactory ServiceScopeFactory { get; }
protected AbpTimer Timer { get; }
protected AbpAsyncTimer Timer { get; }
protected AsyncPeriodicBackgroundWorkerBase(
AbpTimer timer,
AbpAsyncTimer timer,
IServiceScopeFactory serviceScopeFactory)
{
ServiceScopeFactory = serviceScopeFactory;
Timer = timer;
Timer.Elapsed += Timer_Elapsed;
Timer.Elapsed = Timer_Elapsed;
}
public async override Task StartAsync(CancellationToken cancellationToken = default)
@ -34,23 +34,24 @@ namespace Volo.Abp.BackgroundWorkers
await base.StopAsync(cancellationToken);
}
private void Timer_Elapsed(object sender, System.EventArgs e)
private async Task Timer_Elapsed(AbpAsyncTimer timer)
{
await DoWorkAsync();
}
private async Task DoWorkAsync()
{
using (var scope = ServiceScopeFactory.CreateScope())
{
try
{
AsyncHelper.RunSync(
() => DoWorkAsync(new PeriodicBackgroundWorkerContext(scope.ServiceProvider))
);
await DoWorkAsync(new PeriodicBackgroundWorkerContext(scope.ServiceProvider));
}
catch (Exception ex)
{
AsyncHelper.RunSync(
() => scope.ServiceProvider
.GetRequiredService<IExceptionNotifier>()
.NotifyAsync(new ExceptionNotificationContext(ex))
);
await scope.ServiceProvider
.GetRequiredService<IExceptionNotifier>()
.NotifyAsync(new ExceptionNotificationContext(ex));
Logger.LogException(ex);
}

6
framework/src/Volo.Abp.BackgroundWorkers/Volo/Abp/BackgroundWorkers/PeriodicBackgroundWorkerBase.cs

@ -43,14 +43,12 @@ namespace Volo.Abp.BackgroundWorkers
{
try
{
DoWork(new PeriodicBackgroundWorkerContext(scope.ServiceProvider));
}
catch (Exception ex)
{
scope.ServiceProvider
.GetRequiredService<IExceptionNotifier>()
.NotifyAsync(new ExceptionNotificationContext(ex));
var exceptionNotifier = scope.ServiceProvider.GetRequiredService<IExceptionNotifier>();
AsyncHelper.RunSync(() => exceptionNotifier.NotifyAsync(new ExceptionNotificationContext(ex)));
Logger.LogException(ex);
}

2
framework/src/Volo.Abp.Caching/Volo/Abp/Caching/DistributedCache.cs

@ -1029,7 +1029,7 @@ namespace Volo.Abp.Caching
protected virtual void HandleException(Exception ex)
{
AsyncHelper.RunSync(() => HandleExceptionAsync(ex));
_ = HandleExceptionAsync(ex);
}
protected virtual async Task HandleExceptionAsync(Exception ex)

12
framework/src/Volo.Abp.Cli/Volo/Abp/Cli/Program.cs

@ -3,13 +3,13 @@ using Microsoft.Extensions.DependencyInjection;
using Serilog;
using Serilog.Events;
using System.IO;
using Volo.Abp.Threading;
using System.Threading.Tasks;
namespace Volo.Abp.Cli
{
public class Program
{
private static void Main(string[] args)
private static async Task Main(string[] args)
{
Console.OutputEncoding = System.Text.Encoding.UTF8;
@ -37,11 +37,9 @@ namespace Volo.Abp.Cli
{
application.Initialize();
AsyncHelper.RunSync(
() => application.ServiceProvider
.GetRequiredService<CliService>()
.RunAsync(args)
);
await application.ServiceProvider
.GetRequiredService<CliService>()
.RunAsync(args);
application.Shutdown();
}

51
framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs

@ -20,7 +20,7 @@ namespace Volo.Abp.RabbitMQ
protected IExceptionNotifier ExceptionNotifier { get; }
protected AbpTimer Timer { get; }
protected AbpAsyncTimer Timer { get; }
protected ExchangeDeclareConfiguration Exchange { get; private set; }
@ -38,7 +38,7 @@ namespace Volo.Abp.RabbitMQ
public RabbitMqMessageConsumer(
IConnectionPool connectionPool,
AbpTimer timer,
AbpAsyncTimer timer,
IExceptionNotifier exceptionNotifier)
{
ConnectionPool = connectionPool;
@ -50,7 +50,7 @@ namespace Volo.Abp.RabbitMQ
Callbacks = new ConcurrentBag<Func<IModel, BasicDeliverEventArgs, Task>>();
Timer.Period = 5000; //5 sec.
Timer.Elapsed += Timer_Elapsed;
Timer.Elapsed = Timer_Elapsed;
Timer.RunOnStart = true;
}
@ -77,7 +77,7 @@ namespace Volo.Abp.RabbitMQ
await TrySendQueueBindCommandsAsync();
}
protected virtual void TrySendQueueBindCommands()
protected virtual async Task TrySendQueueBindCommandsAsync()
{
try
{
@ -119,40 +119,33 @@ namespace Volo.Abp.RabbitMQ
catch (Exception ex)
{
Logger.LogException(ex, LogLevel.Warning);
AsyncHelper.RunSync(() => ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning));
await ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning);
}
}
protected virtual Task TrySendQueueBindCommandsAsync()
{
TrySendQueueBindCommands();
return Task.CompletedTask;
}
public virtual void OnMessageReceived(Func<IModel, BasicDeliverEventArgs, Task> callback)
{
Callbacks.Add(callback);
}
protected virtual void Timer_Elapsed(object sender, EventArgs e)
protected virtual async Task Timer_Elapsed(AbpAsyncTimer timer)
{
if (Channel == null || Channel.IsOpen == false)
{
TryCreateChannel();
TrySendQueueBindCommands();
await TryCreateChannelAsync();
await TrySendQueueBindCommandsAsync();
}
}
protected virtual void TryCreateChannel()
protected virtual async Task TryCreateChannelAsync()
{
DisposeChannel();
await DisposeChannelAsync();
try
{
var channel = ConnectionPool
.Get(ConnectionName)
.CreateModel();
channel.ExchangeDeclare(
exchange: Exchange.ExchangeName,
type: Exchange.Type,
@ -172,7 +165,7 @@ namespace Volo.Abp.RabbitMQ
var consumer = new EventingBasicConsumer(channel);
consumer.Received += async (model, basicDeliverEventArgs) =>
{
await HandleIncomingMessage(channel, basicDeliverEventArgs);
await HandleIncomingMessageAsync(channel, basicDeliverEventArgs);
};
channel.BasicConsume(
@ -186,11 +179,11 @@ namespace Volo.Abp.RabbitMQ
catch (Exception ex)
{
Logger.LogException(ex, LogLevel.Warning);
AsyncHelper.RunSync(() => ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning));
await ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning);
}
}
protected virtual async Task HandleIncomingMessage(IModel channel, BasicDeliverEventArgs basicDeliverEventArgs)
protected virtual async Task HandleIncomingMessageAsync(IModel channel, BasicDeliverEventArgs basicDeliverEventArgs)
{
try
{
@ -208,6 +201,24 @@ namespace Volo.Abp.RabbitMQ
}
}
protected virtual async Task DisposeChannelAsync()
{
if (Channel == null)
{
return;
}
try
{
Channel.Dispose();
}
catch (Exception ex)
{
Logger.LogException(ex, LogLevel.Warning);
await ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning);
}
}
protected virtual void DisposeChannel()
{
if (Channel == null)

127
framework/src/Volo.Abp.Threading/Volo/Abp/Threading/AbpAsyncTimer.cs

@ -0,0 +1,127 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Volo.Abp.DependencyInjection;
using Volo.Abp.ExceptionHandling;
namespace Volo.Abp.Threading
{
/// <summary>
/// A robust timer implementation that ensures no overlapping occurs. It waits exactly specified <see cref="Period"/> between ticks.
/// </summary>
public class AbpAsyncTimer : ITransientDependency
{
/// <summary>
/// This func is raised periodically according to Period of Timer.
/// </summary>
public Func<AbpAsyncTimer, Task> Elapsed = _ => Task.CompletedTask;
/// <summary>
/// Task period of timer (as milliseconds).
/// </summary>
public int Period { get; set; }
/// <summary>
/// Indicates whether timer raises Elapsed event on Start method of Timer for once.
/// Default: False.
/// </summary>
public bool RunOnStart { get; set; }
public ILogger<AbpAsyncTimer> Logger { get; set; }
public IExceptionNotifier ExceptionNotifier { get; set; }
private readonly Timer _taskTimer;
private volatile bool _performingTasks;
private volatile bool _isRunning;
public AbpAsyncTimer()
{
ExceptionNotifier = NullExceptionNotifier.Instance;
Logger = NullLogger<AbpAsyncTimer>.Instance;
_taskTimer = new Timer(
TimerCallBack,
null,
Timeout.Infinite,
Timeout.Infinite
);
}
public void Start(CancellationToken cancellationToken = default)
{
if (Period <= 0)
{
throw new AbpException("Period should be set before starting the timer!");
}
lock (_taskTimer)
{
_taskTimer.Change(RunOnStart ? 0 : Period, Timeout.Infinite);
_isRunning = true;
}
}
public void Stop(CancellationToken cancellationToken = default)
{
lock (_taskTimer)
{
_taskTimer.Change(Timeout.Infinite, Timeout.Infinite);
while (_performingTasks)
{
Monitor.Wait(_taskTimer);
}
_isRunning = false;
}
}
/// <summary>
/// This method is called by _taskTimer.
/// </summary>
/// <param name="state">Not used argument</param>
private void TimerCallBack(object state)
{
lock (_taskTimer)
{
if (!_isRunning || _performingTasks)
{
return;
}
_taskTimer.Change(Timeout.Infinite, Timeout.Infinite);
_performingTasks = true;
}
_ = Timer_Elapsed();
}
private async Task Timer_Elapsed()
{
try
{
await Elapsed(this);
}
catch(Exception ex)
{
Logger.LogException(ex);
await ExceptionNotifier.NotifyAsync(ex);
}
finally
{
lock (_taskTimer)
{
_performingTasks = false;
if (_isRunning)
{
_taskTimer.Change(Period, Timeout.Infinite);
}
Monitor.Pulse(_taskTimer);
}
}
}
}
}

2
framework/src/Volo.Abp.Threading/Volo/Abp/Threading/AbpTimer.cs

@ -118,4 +118,4 @@ namespace Volo.Abp.Threading
}
}
}
}
}

4
framework/test/Volo.Abp.BlobStoring.Aws.Tests/Volo/Abp/BlobStoring/Aws/AbpBlobStoringAwsTestModule.cs

@ -69,7 +69,7 @@ namespace Volo.Abp.BlobStoring.Aws
private async Task DeleteBucketAsync(ApplicationShutdownContext context)
{
var amazonS3Client = await context.ServiceProvider.GetService<IAmazonS3ClientFactory>()
var amazonS3Client = await context.ServiceProvider.GetRequiredService<IAmazonS3ClientFactory>()
.GetAmazonS3Client(_configuration);
if (await AmazonS3Util.DoesS3BucketExistV2Async(amazonS3Client, _randomContainerName))
@ -89,4 +89,4 @@ namespace Volo.Abp.BlobStoring.Aws
}
}
}
}
}

7
modules/identityserver/src/Volo.Abp.IdentityServer.Domain/Volo/Abp/IdentityServer/Tokens/TokenCleanupBackgroundWorker.cs

@ -11,11 +11,11 @@ namespace Volo.Abp.IdentityServer.Tokens
protected TokenCleanupOptions Options { get; }
public TokenCleanupBackgroundWorker(
AbpTimer timer,
AbpAsyncTimer timer,
IServiceScopeFactory serviceScopeFactory,
IOptions<TokenCleanupOptions> options)
: base(
timer,
timer,
serviceScopeFactory)
{
Options = options.Value;
@ -27,8 +27,7 @@ namespace Volo.Abp.IdentityServer.Tokens
await workerContext
.ServiceProvider
.GetRequiredService<TokenCleanupService>()
.CleanAsync()
;
.CleanAsync();
}
}
}

Loading…
Cancel
Save