|
@ -5,21 +5,20 @@ public class EventBusServiceBus : IEventBus, IDisposable |
|
|
private readonly IServiceBusPersisterConnection _serviceBusPersisterConnection; |
|
|
private readonly IServiceBusPersisterConnection _serviceBusPersisterConnection; |
|
|
private readonly ILogger<EventBusServiceBus> _logger; |
|
|
private readonly ILogger<EventBusServiceBus> _logger; |
|
|
private readonly IEventBusSubscriptionsManager _subsManager; |
|
|
private readonly IEventBusSubscriptionsManager _subsManager; |
|
|
private readonly ILifetimeScope _autofac; |
|
|
|
|
|
|
|
|
private readonly IServiceScopeFactory _serviceScopeFactory; |
|
|
private readonly string _topicName = "eshop_event_bus"; |
|
|
private readonly string _topicName = "eshop_event_bus"; |
|
|
private readonly string _subscriptionName; |
|
|
private readonly string _subscriptionName; |
|
|
private ServiceBusSender _sender; |
|
|
private ServiceBusSender _sender; |
|
|
private ServiceBusProcessor _processor; |
|
|
private ServiceBusProcessor _processor; |
|
|
private readonly string AUTOFAC_SCOPE_NAME = "eshop_event_bus"; |
|
|
|
|
|
private const string INTEGRATION_EVENT_SUFFIX = "IntegrationEvent"; |
|
|
private const string INTEGRATION_EVENT_SUFFIX = "IntegrationEvent"; |
|
|
|
|
|
|
|
|
public EventBusServiceBus(IServiceBusPersisterConnection serviceBusPersisterConnection, |
|
|
public EventBusServiceBus(IServiceBusPersisterConnection serviceBusPersisterConnection, |
|
|
ILogger<EventBusServiceBus> logger, IEventBusSubscriptionsManager subsManager, ILifetimeScope autofac, string subscriptionClientName) |
|
|
|
|
|
|
|
|
ILogger<EventBusServiceBus> logger, IEventBusSubscriptionsManager subsManager, IServiceScopeFactory serviceScopeFactory, string subscriptionClientName) |
|
|
{ |
|
|
{ |
|
|
_serviceBusPersisterConnection = serviceBusPersisterConnection; |
|
|
_serviceBusPersisterConnection = serviceBusPersisterConnection; |
|
|
_logger = logger ?? throw new ArgumentNullException(nameof(logger)); |
|
|
_logger = logger ?? throw new ArgumentNullException(nameof(logger)); |
|
|
_subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager(); |
|
|
_subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager(); |
|
|
_autofac = autofac; |
|
|
|
|
|
|
|
|
_serviceScopeFactory = serviceScopeFactory; |
|
|
_subscriptionName = subscriptionClientName; |
|
|
_subscriptionName = subscriptionClientName; |
|
|
_sender = _serviceBusPersisterConnection.TopicClient.CreateSender(_topicName); |
|
|
_sender = _serviceBusPersisterConnection.TopicClient.CreateSender(_topicName); |
|
|
ServiceBusProcessorOptions options = new ServiceBusProcessorOptions { MaxConcurrentCalls = 10, AutoCompleteMessages = false }; |
|
|
ServiceBusProcessorOptions options = new ServiceBusProcessorOptions { MaxConcurrentCalls = 10, AutoCompleteMessages = false }; |
|
@ -155,14 +154,14 @@ public class EventBusServiceBus : IEventBus, IDisposable |
|
|
var processed = false; |
|
|
var processed = false; |
|
|
if (_subsManager.HasSubscriptionsForEvent(eventName)) |
|
|
if (_subsManager.HasSubscriptionsForEvent(eventName)) |
|
|
{ |
|
|
{ |
|
|
using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME)) |
|
|
|
|
|
|
|
|
using (var scope = _serviceScopeFactory.CreateScope()) |
|
|
{ |
|
|
{ |
|
|
var subscriptions = _subsManager.GetHandlersForEvent(eventName); |
|
|
var subscriptions = _subsManager.GetHandlersForEvent(eventName); |
|
|
foreach (var subscription in subscriptions) |
|
|
foreach (var subscription in subscriptions) |
|
|
{ |
|
|
{ |
|
|
if (subscription.IsDynamic) |
|
|
if (subscription.IsDynamic) |
|
|
{ |
|
|
{ |
|
|
var handler = scope.ResolveOptional(subscription.HandlerType) as IDynamicIntegrationEventHandler; |
|
|
|
|
|
|
|
|
var handler = scope.ServiceProvider.GetService(subscription.HandlerType) as IDynamicIntegrationEventHandler; |
|
|
if (handler == null) continue; |
|
|
if (handler == null) continue; |
|
|
|
|
|
|
|
|
using dynamic eventData = JsonDocument.Parse(message); |
|
|
using dynamic eventData = JsonDocument.Parse(message); |
|
@ -170,7 +169,7 @@ public class EventBusServiceBus : IEventBus, IDisposable |
|
|
} |
|
|
} |
|
|
else |
|
|
else |
|
|
{ |
|
|
{ |
|
|
var handler = scope.ResolveOptional(subscription.HandlerType); |
|
|
|
|
|
|
|
|
var handler = scope.ServiceProvider.GetService(subscription.HandlerType); |
|
|
if (handler == null) continue; |
|
|
if (handler == null) continue; |
|
|
var eventType = _subsManager.GetEventTypeByName(eventName); |
|
|
var eventType = _subsManager.GetEventTypeByName(eventName); |
|
|
var integrationEvent = JsonSerializer.Deserialize(message, eventType); |
|
|
var integrationEvent = JsonSerializer.Deserialize(message, eventType); |
|
|