[FA-4] Adds an event bus infrastructure, a RabbitMQ implementation and rewires existing mutations on NovelService to utilize it.
This commit is contained in:
gamer147
@@ -0,0 +1,25 @@
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus;
|
||||
|
||||
public class EventBusBuilder<TEventBus> where TEventBus : class, IEventBus
|
||||
{
|
||||
private readonly IServiceCollection _services;
|
||||
private readonly SubscriptionManager _subscriptionManager;
|
||||
|
||||
public EventBusBuilder(IServiceCollection services)
|
||||
{
|
||||
_services = services;
|
||||
_services.AddSingleton<IEventBus, TEventBus>();
|
||||
|
||||
_subscriptionManager = new SubscriptionManager();
|
||||
_services.AddSingleton<SubscriptionManager>(_subscriptionManager);
|
||||
}
|
||||
|
||||
public EventBusBuilder<TEventBus> Subscribe<TEvent, TEventHandler>() where TEvent : IntegrationEvent where TEventHandler : class, IIntegrationEventHandler<TEvent>
|
||||
{
|
||||
_services.AddKeyedTransient<IIntegrationEventHandler, TEventHandler>(typeof(TEvent).Name);
|
||||
_subscriptionManager.RegisterSubscription<TEvent>();
|
||||
return this;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus;
|
||||
|
||||
public static class EventBusExtensions
|
||||
{
|
||||
public static EventBusBuilder<TEventBus> AddEventBus<TEventBus>(this IServiceCollection services)
|
||||
where TEventBus : class, IEventBus
|
||||
{
|
||||
return new EventBusBuilder<TEventBus>(services);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus;
|
||||
|
||||
public interface IEventBus
|
||||
{
|
||||
Task Publish<TEvent>(TEvent integrationEvent) where TEvent : IntegrationEvent;
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus;
|
||||
|
||||
public interface IIntegrationEventHandler<in TEvent> : IIntegrationEventHandler where TEvent : IntegrationEvent
|
||||
{
|
||||
Task Handle(TEvent @event);
|
||||
Task IIntegrationEventHandler.Handle(IntegrationEvent @event) => Handle((TEvent)@event);
|
||||
}
|
||||
|
||||
public interface IIntegrationEventHandler
|
||||
{
|
||||
Task Handle(IntegrationEvent @event);
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
using RabbitMQ.Client;
|
||||
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus.Implementations;
|
||||
|
||||
public class RabbitMQConnectionProvider
|
||||
{
|
||||
private readonly IConnectionFactory _connectionFactory;
|
||||
|
||||
private IConnection Connection { get; set; }
|
||||
private IChannel DefaultChannel { get; set; }
|
||||
|
||||
public RabbitMQConnectionProvider(IConnectionFactory connectionFactory)
|
||||
{
|
||||
_connectionFactory = connectionFactory;
|
||||
}
|
||||
|
||||
public async Task<IConnection> GetConnectionAsync()
|
||||
{
|
||||
if (Connection == null)
|
||||
{
|
||||
Connection = await _connectionFactory.CreateConnectionAsync();
|
||||
}
|
||||
|
||||
return Connection;
|
||||
}
|
||||
|
||||
public async Task<IChannel> GetDefaultChannelAsync()
|
||||
{
|
||||
if (DefaultChannel == null)
|
||||
{
|
||||
DefaultChannel = await (await GetConnectionAsync()).CreateChannelAsync();
|
||||
}
|
||||
return DefaultChannel;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Newtonsoft.Json;
|
||||
using NodaTime;
|
||||
using NodaTime.Serialization.JsonNet;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus.Implementations;
|
||||
|
||||
public class RabbitMQEventBus : IEventBus, IHostedService
|
||||
{
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
private readonly RabbitMQConnectionProvider _connectionProvider;
|
||||
private readonly RabbitMQOptions _options;
|
||||
private readonly SubscriptionManager _subscriptionManager;
|
||||
private readonly ILogger<RabbitMQEventBus> _logger;
|
||||
|
||||
private readonly JsonSerializerSettings _jsonSerializerSettings;
|
||||
|
||||
private const string ExchangeName = "fiction-archive-event-bus";
|
||||
|
||||
public RabbitMQEventBus(IServiceScopeFactory serviceScopeFactory, RabbitMQConnectionProvider connectionProvider, IOptions<RabbitMQOptions> options, SubscriptionManager subscriptionManager, ILogger<RabbitMQEventBus> logger)
|
||||
{
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
_connectionProvider = connectionProvider;
|
||||
_subscriptionManager = subscriptionManager;
|
||||
_logger = logger;
|
||||
_options = options.Value;
|
||||
_jsonSerializerSettings = new JsonSerializerSettings().ConfigureForNodaTime(DateTimeZoneProviders.Tzdb);
|
||||
}
|
||||
|
||||
public async Task Publish<TEvent>(TEvent integrationEvent) where TEvent : IntegrationEvent
|
||||
{
|
||||
var routingKey = typeof(TEvent).Name;
|
||||
var channel = await _connectionProvider.GetDefaultChannelAsync();
|
||||
|
||||
// Set integration event values
|
||||
integrationEvent.CreatedAt = Instant.FromDateTimeUtc(DateTime.UtcNow);
|
||||
integrationEvent.EventId = Guid.NewGuid();
|
||||
|
||||
var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(integrationEvent));
|
||||
await channel.BasicPublishAsync(ExchangeName, routingKey, true, body);
|
||||
_logger.LogInformation("Published event {EventName}", routingKey);
|
||||
}
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_ = Task.Factory.StartNew(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
var channel = await _connectionProvider.GetDefaultChannelAsync();
|
||||
await channel.ExchangeDeclareAsync(ExchangeName, ExchangeType.Direct,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
await channel.QueueDeclareAsync(_options.ClientIdentifier, true, false, false,
|
||||
cancellationToken: cancellationToken);
|
||||
var consumer = new AsyncEventingBasicConsumer(channel);
|
||||
consumer.ReceivedAsync += (sender, @event) =>
|
||||
{
|
||||
return OnReceivedEvent(sender, @event, channel);
|
||||
};
|
||||
|
||||
foreach (var subscription in _subscriptionManager.Subscriptions)
|
||||
{
|
||||
await channel.QueueBindAsync(_options.ClientIdentifier, ExchangeName, subscription.Key,
|
||||
cancellationToken: cancellationToken);
|
||||
_logger.LogInformation("Subscribed to {SubscriptionKey}", subscription.Key);
|
||||
}
|
||||
|
||||
await channel.BasicConsumeAsync(_options.ClientIdentifier, false, consumer, cancellationToken: cancellationToken);
|
||||
|
||||
_logger.LogInformation("RabbitMQ EventBus started.");
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError(e, "An error occurred while starting the RabbitMQ EventBus");
|
||||
}
|
||||
}, cancellationToken);
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private async Task OnReceivedEvent(object sender, BasicDeliverEventArgs @event, IChannel channel)
|
||||
{
|
||||
var eventName = @event.RoutingKey;
|
||||
_logger.LogInformation("Received event {EventName}", eventName);
|
||||
try
|
||||
{
|
||||
if (!_subscriptionManager.Subscriptions.ContainsKey(eventName))
|
||||
{
|
||||
_logger.LogWarning("Received event without subscription entry.");
|
||||
return;
|
||||
}
|
||||
|
||||
var eventBody = Encoding.UTF8.GetString(@event.Body.Span);
|
||||
var eventObject = JsonConvert.DeserializeObject(eventBody, _subscriptionManager.Subscriptions[eventName], _jsonSerializerSettings) as IntegrationEvent;
|
||||
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
|
||||
foreach (var service in scope.ServiceProvider.GetKeyedServices<IIntegrationEventHandler>(eventName))
|
||||
{
|
||||
await service.Handle(eventObject);
|
||||
}
|
||||
_logger.LogInformation("Finished handling event with name {EventName}", eventName);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError(e, "An error occurred while handling an event.");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await channel.BasicAckAsync(@event.DeliveryTag, false);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Options;
|
||||
using RabbitMQ.Client;
|
||||
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus.Implementations;
|
||||
|
||||
public static class RabbitMQExtensions
|
||||
{
|
||||
public static EventBusBuilder<RabbitMQEventBus> AddRabbitMQ(this IServiceCollection services, Action<RabbitMQOptions> configure)
|
||||
{
|
||||
services.Configure(configure);
|
||||
services.AddSingleton<IConnectionFactory, ConnectionFactory>(provider =>
|
||||
{
|
||||
var options = provider.GetService<IOptions<RabbitMQOptions>>();
|
||||
ConnectionFactory factory = new ConnectionFactory();
|
||||
factory.Uri = new Uri(options.Value.ConnectionString);
|
||||
return factory;
|
||||
});
|
||||
services.AddSingleton<RabbitMQConnectionProvider>();
|
||||
services.AddHostedService<RabbitMQEventBus>();
|
||||
return services.AddEventBus<RabbitMQEventBus>();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus.Implementations;
|
||||
|
||||
public class RabbitMQOptions
|
||||
{
|
||||
public string ConnectionString { get; set; }
|
||||
public string ClientIdentifier { get; set; }
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
using NodaTime;
|
||||
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus;
|
||||
|
||||
public abstract class IntegrationEvent
|
||||
{
|
||||
public Guid EventId { get; set; }
|
||||
public Instant CreatedAt { get; set; }
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
namespace FictionArchive.Service.Shared.Services.EventBus;
|
||||
|
||||
public class SubscriptionManager
|
||||
{
|
||||
public Dictionary<string, Type> Subscriptions { get; } = new Dictionary<string, Type>();
|
||||
|
||||
public void RegisterSubscription<TEvent>()
|
||||
{
|
||||
Subscriptions.Add(typeof(TEvent).Name, typeof(TEvent));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user