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

Make ServiceBusFactory thread-safe

Browse source
This commit is contained in:
Sipke Schoorstra 2021-11-02 20:57:42 +01:00
parent bf87a2a699
commit feff07bda5
1 changed files with 18 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
src/core/Elsa.Core/Services/Messaging/ServiceBusFactory.cs
View file

@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using Elsa.Options;
using Elsa.Serialization;
using Microsoft.Extensions.Logging;
@ -20,6 +21,7 @@ namespace Elsa.Services.Messaging
private readonly IDictionary<string, IBus> _serviceBuses = new Dictionary<string, IBus>();
private readonly IDictionary<Type, string> _messageTypeQueueDictionary = new Dictionary<Type, string>();
private readonly DependencyInjectionHandlerActivator _handlerActivator;
private readonly SemaphoreSlim _semaphore = new(1);
public ServiceBusFactory(ElsaOptions elsaOptions, ILoggerFactory loggerFactory, IServiceProvider serviceProvider)
{
@ -67,15 +69,24 @@ namespace Elsa.Services.Messaging
private IBus GetOrCreateServiceBus(Type messageType, string? queueName)
{
queueName ??= _messageTypeQueueDictionary[messageType];
if (!_serviceBuses.TryGetValue(queueName, out var bus))
_semaphore.Wait();
try
{
bus = ConfigureServiceBus(new[] { messageType }, queueName);
_serviceBuses[queueName] = bus;
}
queueName ??= _messageTypeQueueDictionary[messageType];
return bus;
if (!_serviceBuses.TryGetValue(queueName, out var bus))
{
bus = ConfigureServiceBus(new[] { messageType }, queueName);
_serviceBuses[queueName] = bus;
}
return bus;
}
finally
{
_semaphore.Release();
}
}
private string PrefixQueueName(string name) => $"{_elsaOptions.ServiceBusOptions.QueuePrefix}{name}";