ayan.poddar
/
eShopOnContainers
1
0
Fork 0
Code Issues 0 Pull Requests 0 Projects 0 Releases 19 Wiki Activity
Browse Source

Moved using statements to globalusing for EventBusServiceBus

pull/1770/head
Sumit Ghosh 3 years ago
parent
409fab7521
commit
03b80d037e
3 changed files with 196 additions and 216 deletions
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +42
    -46
      src/BuildingBlocks/EventBus/EventBusServiceBus/DefaultServiceBusPersisterConnection.cs
  2. +148
    -160
      src/BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.cs
  3. +6
    -10
      src/BuildingBlocks/EventBus/EventBusServiceBus/IServiceBusPersisterConnection.cs

+ 42
- 46
src/BuildingBlocks/EventBus/EventBusServiceBus/DefaultServiceBusPersisterConnection.cs View File

@ -1,68 +1,64 @@
using Microsoft.Azure.ServiceBus;
using System;
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus;
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus
public class DefaultServiceBusPersisterConnection : IServiceBusPersisterConnection
{ {
public class DefaultServiceBusPersisterConnection : IServiceBusPersisterConnection
{
private readonly ServiceBusConnectionStringBuilder _serviceBusConnectionStringBuilder;
private readonly string _subscriptionClientName;
private SubscriptionClient _subscriptionClient;
private ITopicClient _topicClient;
private readonly ServiceBusConnectionStringBuilder _serviceBusConnectionStringBuilder;
private readonly string _subscriptionClientName;
private SubscriptionClient _subscriptionClient;
private ITopicClient _topicClient;
bool _disposed;
bool _disposed;
public DefaultServiceBusPersisterConnection(ServiceBusConnectionStringBuilder serviceBusConnectionStringBuilder,
string subscriptionClientName)
{
_serviceBusConnectionStringBuilder = serviceBusConnectionStringBuilder ??
throw new ArgumentNullException(nameof(serviceBusConnectionStringBuilder));
_subscriptionClientName = subscriptionClientName;
_subscriptionClient = new SubscriptionClient(_serviceBusConnectionStringBuilder, subscriptionClientName);
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
}
public DefaultServiceBusPersisterConnection(ServiceBusConnectionStringBuilder serviceBusConnectionStringBuilder,
string subscriptionClientName)
{
_serviceBusConnectionStringBuilder = serviceBusConnectionStringBuilder ??
throw new ArgumentNullException(nameof(serviceBusConnectionStringBuilder));
_subscriptionClientName = subscriptionClientName;
_subscriptionClient = new SubscriptionClient(_serviceBusConnectionStringBuilder, subscriptionClientName);
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
}
public ITopicClient TopicClient
public ITopicClient TopicClient
{
get
{ {
get
if (_topicClient.IsClosedOrClosing)
{ {
if (_topicClient.IsClosedOrClosing)
{
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
}
return _topicClient;
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
} }
return _topicClient;
} }
}
public ISubscriptionClient SubscriptionClient
public ISubscriptionClient SubscriptionClient
{
get
{ {
get
if (_subscriptionClient.IsClosedOrClosing)
{ {
if (_subscriptionClient.IsClosedOrClosing)
{
_subscriptionClient = new SubscriptionClient(_serviceBusConnectionStringBuilder, _subscriptionClientName);
}
return _subscriptionClient;
_subscriptionClient = new SubscriptionClient(_serviceBusConnectionStringBuilder, _subscriptionClientName);
} }
return _subscriptionClient;
} }
}
public ServiceBusConnectionStringBuilder ServiceBusConnectionStringBuilder => _serviceBusConnectionStringBuilder;
public ServiceBusConnectionStringBuilder ServiceBusConnectionStringBuilder => _serviceBusConnectionStringBuilder;
public ITopicClient CreateModel()
public ITopicClient CreateModel()
{
if (_topicClient.IsClosedOrClosing)
{ {
if (_topicClient.IsClosedOrClosing)
{
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
}
return _topicClient;
_topicClient = new TopicClient(_serviceBusConnectionStringBuilder, RetryPolicy.Default);
} }
public void Dispose()
{
if (_disposed) return;
return _topicClient;
}
_disposed = true;
}
public void Dispose()
{
if (_disposed) return;
_disposed = true;
} }
} }

+ 148
- 160
src/BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.cs View File

@ -1,203 +1,191 @@
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus;
public class EventBusServiceBus : IEventBus
{ {
using Autofac;
using Microsoft.Azure.ServiceBus;
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
using Microsoft.Extensions.Logging;
using System;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;
public class EventBusServiceBus : IEventBus
private readonly IServiceBusPersisterConnection _serviceBusPersisterConnection;
private readonly ILogger<EventBusServiceBus> _logger;
private readonly IEventBusSubscriptionsManager _subsManager;
private readonly ILifetimeScope _autofac;
private readonly string AUTOFAC_SCOPE_NAME = "eshop_event_bus";
private const string INTEGRATION_EVENT_SUFFIX = "IntegrationEvent";
public EventBusServiceBus(IServiceBusPersisterConnection serviceBusPersisterConnection,
ILogger<EventBusServiceBus> logger, IEventBusSubscriptionsManager subsManager, ILifetimeScope autofac)
{ {
private readonly IServiceBusPersisterConnection _serviceBusPersisterConnection;
private readonly ILogger<EventBusServiceBus> _logger;
private readonly IEventBusSubscriptionsManager _subsManager;
private readonly ILifetimeScope _autofac;
private readonly string AUTOFAC_SCOPE_NAME = "eshop_event_bus";
private const string INTEGRATION_EVENT_SUFFIX = "IntegrationEvent";
public EventBusServiceBus(IServiceBusPersisterConnection serviceBusPersisterConnection,
ILogger<EventBusServiceBus> logger, IEventBusSubscriptionsManager subsManager, ILifetimeScope autofac)
{
_serviceBusPersisterConnection = serviceBusPersisterConnection;
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
_autofac = autofac;
RemoveDefaultRule();
RegisterSubscriptionClientMessageHandler();
}
public void Publish(IntegrationEvent @event)
{
var eventName = @event.GetType().Name.Replace(INTEGRATION_EVENT_SUFFIX, "");
var jsonMessage = JsonSerializer.Serialize(@event);
var body = Encoding.UTF8.GetBytes(jsonMessage);
var message = new Message
{
MessageId = Guid.NewGuid().ToString(),
Body = body,
Label = eventName,
};
_serviceBusPersisterConnection.TopicClient.SendAsync(message)
.GetAwaiter()
.GetResult();
}
_serviceBusPersisterConnection = serviceBusPersisterConnection;
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
_autofac = autofac;
public void SubscribeDynamic<TH>(string eventName)
where TH : IDynamicIntegrationEventHandler
{
_logger.LogInformation("Subscribing to dynamic event {EventName} with {EventHandler}", eventName, typeof(TH).Name);
RemoveDefaultRule();
RegisterSubscriptionClientMessageHandler();
}
_subsManager.AddDynamicSubscription<TH>(eventName);
}
public void Publish(IntegrationEvent @event)
{
var eventName = @event.GetType().Name.Replace(INTEGRATION_EVENT_SUFFIX, "");
var jsonMessage = JsonSerializer.Serialize(@event);
var body = Encoding.UTF8.GetBytes(jsonMessage);
public void Subscribe<T, TH>()
where T : IntegrationEvent
where TH : IIntegrationEventHandler<T>
var message = new Message
{ {
var eventName = typeof(T).Name.Replace(INTEGRATION_EVENT_SUFFIX, "");
MessageId = Guid.NewGuid().ToString(),
Body = body,
Label = eventName,
};
_serviceBusPersisterConnection.TopicClient.SendAsync(message)
.GetAwaiter()
.GetResult();
}
var containsKey = _subsManager.HasSubscriptionsForEvent<T>();
if (!containsKey)
{
try
{
_serviceBusPersisterConnection.SubscriptionClient.AddRuleAsync(new RuleDescription
{
Filter = new CorrelationFilter { Label = eventName },
Name = eventName
}).GetAwaiter().GetResult();
}
catch (ServiceBusException)
{
_logger.LogWarning("The messaging entity {eventName} already exists.", eventName);
}
}
public void SubscribeDynamic<TH>(string eventName)
where TH : IDynamicIntegrationEventHandler
{
_logger.LogInformation("Subscribing to dynamic event {EventName} with {EventHandler}", eventName, typeof(TH).Name);
_logger.LogInformation("Subscribing to event {EventName} with {EventHandler}", eventName, typeof(TH).Name);
_subsManager.AddDynamicSubscription<TH>(eventName);
}
_subsManager.AddSubscription<T, TH>();
}
public void Subscribe<T, TH>()
where T : IntegrationEvent
where TH : IIntegrationEventHandler<T>
{
var eventName = typeof(T).Name.Replace(INTEGRATION_EVENT_SUFFIX, "");
public void Unsubscribe<T, TH>()
where T : IntegrationEvent
where TH : IIntegrationEventHandler<T>
var containsKey = _subsManager.HasSubscriptionsForEvent<T>();
if (!containsKey)
{ {
var eventName = typeof(T).Name.Replace(INTEGRATION_EVENT_SUFFIX, "");
try try
{ {
_serviceBusPersisterConnection
.SubscriptionClient
.RemoveRuleAsync(eventName)
.GetAwaiter()
.GetResult();
_serviceBusPersisterConnection.SubscriptionClient.AddRuleAsync(new RuleDescription
{
Filter = new CorrelationFilter { Label = eventName },
Name = eventName
}).GetAwaiter().GetResult();
} }
catch (MessagingEntityNotFoundException)
catch (ServiceBusException)
{ {
_logger.LogWarning("The messaging entity {eventName} Could not be found.", eventName);
_logger.LogWarning("The messaging entity {eventName} already exists.", eventName);
} }
}
_logger.LogInformation("Unsubscribing from event {EventName}", eventName);
_logger.LogInformation("Subscribing to event {EventName} with {EventHandler}", eventName, typeof(TH).Name);
_subsManager.RemoveSubscription<T, TH>();
}
_subsManager.AddSubscription<T, TH>();
}
public void UnsubscribeDynamic<TH>(string eventName)
where TH : IDynamicIntegrationEventHandler
{
_logger.LogInformation("Unsubscribing from dynamic event {EventName}", eventName);
public void Unsubscribe<T, TH>()
where T : IntegrationEvent
where TH : IIntegrationEventHandler<T>
{
var eventName = typeof(T).Name.Replace(INTEGRATION_EVENT_SUFFIX, "");
_subsManager.RemoveDynamicSubscription<TH>(eventName);
try
{
_serviceBusPersisterConnection
.SubscriptionClient
.RemoveRuleAsync(eventName)
.GetAwaiter()
.GetResult();
} }
public void Dispose()
catch (MessagingEntityNotFoundException)
{ {
_subsManager.Clear();
_logger.LogWarning("The messaging entity {eventName} Could not be found.", eventName);
} }
private void RegisterSubscriptionClientMessageHandler()
{
_serviceBusPersisterConnection.SubscriptionClient.RegisterMessageHandler(
async (message, token) =>
{
var eventName = $"{message.Label}{INTEGRATION_EVENT_SUFFIX}";
var messageData = Encoding.UTF8.GetString(message.Body);
_logger.LogInformation("Unsubscribing from event {EventName}", eventName);
// Complete the message so that it is not received again.
if (await ProcessEvent(eventName, messageData))
{
await _serviceBusPersisterConnection.SubscriptionClient.CompleteAsync(message.SystemProperties.LockToken);
}
},
new MessageHandlerOptions(ExceptionReceivedHandler) { MaxConcurrentCalls = 10, AutoComplete = false });
}
_subsManager.RemoveSubscription<T, TH>();
}
private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
{
var ex = exceptionReceivedEventArgs.Exception;
var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
public void UnsubscribeDynamic<TH>(string eventName)
where TH : IDynamicIntegrationEventHandler
{
_logger.LogInformation("Unsubscribing from dynamic event {EventName}", eventName);
_logger.LogError(ex, "ERROR handling message: {ExceptionMessage} - Context: {@ExceptionContext}", ex.Message, context);
_subsManager.RemoveDynamicSubscription<TH>(eventName);
}
return Task.CompletedTask;
}
public void Dispose()
{
_subsManager.Clear();
}
private void RegisterSubscriptionClientMessageHandler()
{
_serviceBusPersisterConnection.SubscriptionClient.RegisterMessageHandler(
async (message, token) =>
{
var eventName = $"{message.Label}{INTEGRATION_EVENT_SUFFIX}";
var messageData = Encoding.UTF8.GetString(message.Body);
// Complete the message so that it is not received again.
if (await ProcessEvent(eventName, messageData))
{
await _serviceBusPersisterConnection.SubscriptionClient.CompleteAsync(message.SystemProperties.LockToken);
}
},
new MessageHandlerOptions(ExceptionReceivedHandler) { MaxConcurrentCalls = 10, AutoComplete = false });
}
private async Task<bool> ProcessEvent(string eventName, string message)
private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
{
var ex = exceptionReceivedEventArgs.Exception;
var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
_logger.LogError(ex, "ERROR handling message: {ExceptionMessage} - Context: {@ExceptionContext}", ex.Message, context);
return Task.CompletedTask;
}
private async Task<bool> ProcessEvent(string eventName, string message)
{
var processed = false;
if (_subsManager.HasSubscriptionsForEvent(eventName))
{ {
var processed = false;
if (_subsManager.HasSubscriptionsForEvent(eventName))
using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME))
{ {
using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME))
var subscriptions = _subsManager.GetHandlersForEvent(eventName);
foreach (var subscription in subscriptions)
{ {
var subscriptions = _subsManager.GetHandlersForEvent(eventName);
foreach (var subscription in subscriptions)
if (subscription.IsDynamic)
{ {
if (subscription.IsDynamic)
{
var handler = scope.ResolveOptional(subscription.HandlerType) as IDynamicIntegrationEventHandler;
if (handler == null) continue;
var handler = scope.ResolveOptional(subscription.HandlerType) as IDynamicIntegrationEventHandler;
if (handler == null) continue;
using dynamic eventData = JsonDocument.Parse(message);
await handler.Handle(eventData);
}
else
{
var handler = scope.ResolveOptional(subscription.HandlerType);
if (handler == null) continue;
var eventType = _subsManager.GetEventTypeByName(eventName);
var integrationEvent = JsonSerializer.Deserialize(message, eventType);
var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);
await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent });
}
using dynamic eventData = JsonDocument.Parse(message);
await handler.Handle(eventData);
}
else
{
var handler = scope.ResolveOptional(subscription.HandlerType);
if (handler == null) continue;
var eventType = _subsManager.GetEventTypeByName(eventName);
var integrationEvent = JsonSerializer.Deserialize(message, eventType);
var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);
await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent });
} }
} }
processed = true;
} }
return processed;
processed = true;
} }
return processed;
}
private void RemoveDefaultRule()
private void RemoveDefaultRule()
{
try
{ {
try
{
_serviceBusPersisterConnection
.SubscriptionClient
.RemoveRuleAsync(RuleDescription.DefaultRuleName)
.GetAwaiter()
.GetResult();
}
catch (MessagingEntityNotFoundException)
{
_logger.LogWarning("The messaging entity {DefaultRuleName} Could not be found.", RuleDescription.DefaultRuleName);
}
_serviceBusPersisterConnection
.SubscriptionClient
.RemoveRuleAsync(RuleDescription.DefaultRuleName)
.GetAwaiter()
.GetResult();
}
catch (MessagingEntityNotFoundException)
{
_logger.LogWarning("The messaging entity {DefaultRuleName} Could not be found.", RuleDescription.DefaultRuleName);
} }
} }
} }

+ 6
- 10
src/BuildingBlocks/EventBus/EventBusServiceBus/IServiceBusPersisterConnection.cs View File

@ -1,11 +1,7 @@
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus
{
using Microsoft.Azure.ServiceBus;
using System;
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus;
public interface IServiceBusPersisterConnection : IDisposable
{
ITopicClient TopicClient { get; }
ISubscriptionClient SubscriptionClient { get; }
}
}
public interface IServiceBusPersisterConnection : IDisposable
{
ITopicClient TopicClient { get; }
ISubscriptionClient SubscriptionClient { get; }
}

Write Preview
Loading…
Cancel
Save