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

Fix concurrent access to Azure Service Bus workers

Browse source
This commit is contained in:
Sipke Schoorstra 2021-05-28 19:36:16 +02:00
parent be61cf1a51
commit 2d2fa4451f
1 changed files with 14 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
src/activities/Elsa.Activities.AzureServiceBus/Services/ServiceBusQueuesStarter.cs
View file

@ -19,6 +19,7 @@ namespace Elsa.Activities.AzureServiceBus.Services
private readonly IServiceProvider _serviceProvider;
private readonly ILogger _logger;
private readonly ICollection<QueueWorker> _workers;
private readonly SemaphoreSlim _semaphore = new(1);
public ServiceBusQueuesStarter(
IQueueMessageReceiverClientFactory messageReceiverClientFactory,
@ -35,11 +36,20 @@ namespace Elsa.Activities.AzureServiceBus.Services
public async Task CreateWorkersAsync(CancellationToken cancellationToken = default)
{
await DisposeExistingWorkersAsync();
var queueNames = (await GetQueueNamesAsync(cancellationToken).ToListAsync(cancellationToken)).Distinct();
await _semaphore.WaitAsync(cancellationToken);
foreach (var queueName in queueNames)
await CreateAndAddWorkerAsync(queueName, cancellationToken);
try
{
await DisposeExistingWorkersAsync();
var queueNames = (await GetQueueNamesAsync(cancellationToken).ToListAsync(cancellationToken)).Distinct();
foreach (var queueName in queueNames)
await CreateAndAddWorkerAsync(queueName, cancellationToken);
}
finally
{
_semaphore.Release();
}
}
private async Task CreateAndAddWorkerAsync(string queueName, CancellationToken cancellationToken)