SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Update Azure Service Bus queue worker to schedule workflow execution in the background

Browse source 
This to prevent scenarios where workflows take a long time to execute and the message gets re-delivered.
This change ACKS the message right after the workflow task has been queued internally.
This commit is contained in:
Sipke Schoorstra 2020-12-30 12:02:13 +01:00
parent a0c9f16aae
commit 356e0ba0ba
1 changed files with 18 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs
View file

@ -14,14 +14,16 @@ namespace Elsa.Activities.AzureServiceBus.Services
{
private readonly IMessageReceiver _messageReceiver;
private readonly IServiceProvider _serviceProvider;
private readonly IBackgroundWorker _backgroundWorker;
private readonly ILogger _logger;
public QueueWorker(IMessageReceiver messageReceiver, IServiceProvider serviceProvider, ILogger<QueueWorker> logger)
public QueueWorker(IMessageReceiver messageReceiver, IServiceProvider serviceProvider, IBackgroundWorker backgroundWorker, ILogger<QueueWorker> logger)
{
_messageReceiver = messageReceiver;
_serviceProvider = serviceProvider;
_backgroundWorker = backgroundWorker;
_logger = logger;
_messageReceiver.RegisterMessageHandler(OnMessageReceived, new MessageHandlerOptions(ExceptionReceivedHandler)
{
AutoComplete = false,
@ -31,17 +33,21 @@ namespace Elsa.Activities.AzureServiceBus.Services
private async Task OnMessageReceived(Message message, CancellationToken cancellationToken)
{
using (var scope = _serviceProvider.CreateScope())
{
var workflowRunner = scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
await workflowRunner.TriggerWorkflowsAsync<MessageReceivedTrigger>(x => x.QueueName == _messageReceiver.Path && (string.IsNullOrWhiteSpace(x.CorrelationId) || x.CorrelationId == message.CorrelationId), message,
message.CorrelationId, cancellationToken: cancellationToken);
}
await _backgroundWorker.ScheduleTask(async () => await TriggerWorkflowsAsync(message, cancellationToken), cancellationToken);
await _messageReceiver.CompleteAsync(message.SystemProperties.LockToken);
}
private async Task TriggerWorkflowsAsync(Message message, CancellationToken cancellationToken)
{
using var scope = _serviceProvider.CreateScope();
var workflowRunner = scope.ServiceProvider.GetRequiredService<IWorkflowRunner>();
await workflowRunner.TriggerWorkflowsAsync<MessageReceivedTrigger>(
x => x.QueueName == _messageReceiver.Path && (string.IsNullOrWhiteSpace(x.CorrelationId) || x.CorrelationId == message.CorrelationId),
message, message.CorrelationId,
cancellationToken: cancellationToken);
}
private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e)
{
var context = e.ExceptionReceivedContext;
@ -60,13 +66,10 @@ namespace Elsa.Activities.AzureServiceBus.Services
_logger.LogError("- Executing Action: {Action}", context.Action);
break;
}
return Task.CompletedTask;
}
public async ValueTask DisposeAsync()
{
await _messageReceiver.CloseAsync();
}
public async ValueTask DisposeAsync() => await _messageReceiver.CloseAsync();
}
}