Upstream/bitwarden-server
1
0
Fork 0
mirror of https://github.com/bitwarden/server.git synced 2025-02-19 02:21:21 +01:00
Code Issues Projects Releases Wiki Activity

[PM-17562] Refactor existing RabbitMq implementation (#5357)

Browse Source 
* [PM-17562] Refactor existing RabbitMq implementation

* Fixed issues noted in PR review
This commit is contained in:
Brant DeBow 2025-02-04 08:02:43 -06:00 committed by GitHub
parent f1b9bd9a09
commit 3f3da558b6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 162 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/Core/AdminConsole/Services/IEventMessageHandler.cs Normal file
View File

@ -0,0 +1,8 @@
using Bit.Core.Models.Data;
namespace Bit.Core.Services;
public interface IEventMessageHandler
{
Task HandleEventAsync(EventMessage eventMessage);
}

14
src/Core/AdminConsole/Services/Implementations/EventRepositoryHandler.cs Normal file
View File

@ -0,0 +1,14 @@
using Bit.Core.Models.Data;
using Microsoft.Extensions.DependencyInjection;
namespace Bit.Core.Services;
public class EventRepositoryHandler(
[FromKeyedServices("persistent")] IEventWriteService eventWriteService)
: IEventMessageHandler
{
public Task HandleEventAsync(EventMessage eventMessage)
{
return eventWriteService.CreateAsync(eventMessage);
}
}

15
src/Core/AdminConsole/Services/Implementations/RabbitMqEventHttpPostListener.cs → src/Core/AdminConsole/Services/Implementations/HttpPostEventHandler.cs
View File

@ -1,32 +1,25 @@
using System.Net.Http.Json;
using Bit.Core.Models.Data;
using Bit.Core.Settings;
using Microsoft.Extensions.Logging;
namespace Bit.Core.Services;
public class RabbitMqEventHttpPostListener : RabbitMqEventListenerBase
public class HttpPostEventHandler : IEventMessageHandler
{
private readonly HttpClient _httpClient;
private readonly string _httpPostUrl;
private readonly string _queueName;
protected override string QueueName => _queueName;
public const string HttpClientName = "HttpPostEventHandlerHttpClient";
public const string HttpClientName = "EventHttpPostListenerHttpClient";
public RabbitMqEventHttpPostListener(
public HttpPostEventHandler(
IHttpClientFactory httpClientFactory,
ILogger<RabbitMqEventListenerBase> logger,
GlobalSettings globalSettings)
: base(logger, globalSettings)
{
_httpClient = httpClientFactory.CreateClient(HttpClientName);
_httpPostUrl = globalSettings.EventLogging.RabbitMq.HttpPostUrl;
_queueName = globalSettings.EventLogging.RabbitMq.HttpPostQueueName;
}
protected override async Task HandleMessageAsync(EventMessage eventMessage)
public async Task HandleEventAsync(EventMessage eventMessage)
{
var content = JsonContent.Create(eventMessage);
var response = await _httpClient.PostAsync(_httpPostUrl, content);

29
src/Core/AdminConsole/Services/Implementations/RabbitMqEventRepositoryListener.cs
View File

@ -1,29 +0,0 @@
using Bit.Core.Models.Data;
using Bit.Core.Settings;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Bit.Core.Services;
public class RabbitMqEventRepositoryListener : RabbitMqEventListenerBase
{
private readonly IEventWriteService _eventWriteService;
private readonly string _queueName;
protected override string QueueName => _queueName;
public RabbitMqEventRepositoryListener(
[FromKeyedServices("persistent")] IEventWriteService eventWriteService,
ILogger<RabbitMqEventListenerBase> logger,
GlobalSettings globalSettings)
: base(logger, globalSettings)
{
_eventWriteService = eventWriteService;
_queueName = globalSettings.EventLogging.RabbitMq.EventRepositoryQueueName;
}
protected override Task HandleMessageAsync(EventMessage eventMessage)
{
return _eventWriteService.CreateAsync(eventMessage);
}
}

13
src/Core/Services/EventLoggingListenerService.cs Normal file
View File

@ -0,0 +1,13 @@
using Microsoft.Extensions.Hosting;
namespace Bit.Core.Services;
public abstract class EventLoggingListenerService : BackgroundService
{
protected readonly IEventMessageHandler _handler;
protected EventLoggingListenerService(IEventMessageHandler handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
}

27
src/Core/AdminConsole/Services/Implementations/RabbitMqEventListenerBase.cs → src/Core/Services/Implementations/RabbitMqEventListenerService.cs
View File

@ -1,26 +1,26 @@
using System.Text.Json;
using Bit.Core.Models.Data;
using Bit.Core.Settings;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace Bit.Core.Services;
public abstract class RabbitMqEventListenerBase : BackgroundService
public class RabbitMqEventListenerService : EventLoggingListenerService
{
private IChannel _channel;
private IConnection _connection;
private readonly string _exchangeName;
private readonly ConnectionFactory _factory;
private readonly ILogger<RabbitMqEventListenerBase> _logger;
private readonly ILogger<RabbitMqEventListenerService> _logger;
private readonly string _queueName;
protected abstract string QueueName { get; }
protected RabbitMqEventListenerBase(
ILogger<RabbitMqEventListenerBase> logger,
GlobalSettings globalSettings)
public RabbitMqEventListenerService(
IEventMessageHandler handler,
ILogger<RabbitMqEventListenerService> logger,
GlobalSettings globalSettings,
string queueName) : base(handler)
{
_factory = new ConnectionFactory
{
@ -30,6 +30,7 @@ public abstract class RabbitMqEventListenerBase : BackgroundService
};
_exchangeName = globalSettings.EventLogging.RabbitMq.ExchangeName;
_logger = logger;
_queueName = queueName;
}
public override async Task StartAsync(CancellationToken cancellationToken)
@ -38,13 +39,13 @@ public abstract class RabbitMqEventListenerBase : BackgroundService
_channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
await _channel.ExchangeDeclareAsync(exchange: _exchangeName, type: ExchangeType.Fanout, durable: true);
await _channel.QueueDeclareAsync(queue: QueueName,
await _channel.QueueDeclareAsync(queue: _queueName,
durable: true,
exclusive: false,
autoDelete: false,
arguments: null,
cancellationToken: cancellationToken);
await _channel.QueueBindAsync(queue: QueueName,
await _channel.QueueBindAsync(queue: _queueName,
exchange: _exchangeName,
routingKey: string.Empty,
cancellationToken: cancellationToken);
@ -59,7 +60,7 @@ public abstract class RabbitMqEventListenerBase : BackgroundService
try
{
var eventMessage = JsonSerializer.Deserialize<EventMessage>(eventArgs.Body.Span);
await HandleMessageAsync(eventMessage);
await _handler.HandleEventAsync(eventMessage);
}
catch (Exception ex)
{
@ -67,7 +68,7 @@ public abstract class RabbitMqEventListenerBase : BackgroundService
}
};
await _channel.BasicConsumeAsync(QueueName, autoAck: true, consumer: consumer, cancellationToken: stoppingToken);
await _channel.BasicConsumeAsync(_queueName, autoAck: true, consumer: consumer, cancellationToken: stoppingToken);
while (!stoppingToken.IsCancellationRequested)
{
@ -88,6 +89,4 @@ public abstract class RabbitMqEventListenerBase : BackgroundService
_connection.Dispose();
base.Dispose();
}
protected abstract Task HandleMessageAsync(EventMessage eventMessage);
}

1
src/Core/Settings/IGlobalSettings.cs
View File

@ -27,4 +27,5 @@ public interface IGlobalSettings
string DatabaseProvider { get; set; }
GlobalSettings.SqlSettings SqlServer { get; set; }
string DevelopmentDirectory { get; set; }
GlobalSettings.EventLoggingSettings EventLogging { get; set; }
}

19
src/Events/Startup.cs
View File

@ -89,13 +89,26 @@ public class Startup
CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.Password) &&
CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.ExchangeName))
{
services.AddSingleton<EventRepositoryHandler>();
services.AddKeyedSingleton<IEventWriteService, RepositoryEventWriteService>("persistent");
services.AddHostedService<RabbitMqEventRepositoryListener>();
services.AddSingleton<IHostedService>(provider =>
new RabbitMqEventListenerService(
provider.GetRequiredService<EventRepositoryHandler>(),
provider.GetRequiredService<ILogger<RabbitMqEventListenerService>>(),
provider.GetRequiredService<GlobalSettings>(),
globalSettings.EventLogging.RabbitMq.EventRepositoryQueueName));
if (CoreHelpers.SettingHasValue(globalSettings.EventLogging.RabbitMq.HttpPostUrl))
{
services.AddHttpClient(RabbitMqEventHttpPostListener.HttpClientName);
services.AddHostedService<RabbitMqEventHttpPostListener>();
services.AddSingleton<HttpPostEventHandler>();
services.AddHttpClient(HttpPostEventHandler.HttpClientName);
services.AddSingleton<IHostedService>(provider =>
new RabbitMqEventListenerService(
provider.GetRequiredService<HttpPostEventHandler>(),
provider.GetRequiredService<ILogger<RabbitMqEventListenerService>>(),
provider.GetRequiredService<GlobalSettings>(),
globalSettings.EventLogging.RabbitMq.HttpPostQueueName));
}
}
}

3
test/Common/MockedHttpClient/MockedHttpMessageHandler.cs
View File

@ -8,6 +8,8 @@ public class MockedHttpMessageHandler : HttpMessageHandler
{
private readonly List<IHttpRequestMatcher> _matchers = new();
public List<HttpRequestMessage> CapturedRequests { get; } = new List<HttpRequestMessage>();
/// <summary>
/// The fallback handler to use when the request does not match any of the provided matchers.
/// </summary>
@ -16,6 +18,7 @@ public class MockedHttpMessageHandler : HttpMessageHandler
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
CapturedRequests.Add(request);
var matcher = _matchers.FirstOrDefault(x => x.Matches(request));
if (matcher == null)
{

24
test/Core.Test/AdminConsole/Services/EventRepositoryHandlerTests.cs Normal file
View File

@ -0,0 +1,24 @@
using Bit.Core.Models.Data;
using Bit.Core.Services;
using Bit.Test.Common.AutoFixture;
using Bit.Test.Common.AutoFixture.Attributes;
using Bit.Test.Common.Helpers;
using NSubstitute;
using Xunit;
namespace Bit.Core.Test.Services;
[SutProviderCustomize]
public class EventRepositoryHandlerTests
{
[Theory, BitAutoData]
public async Task HandleEventAsync_WritesEventToIEventWriteService(
EventMessage eventMessage,
SutProvider<EventRepositoryHandler> sutProvider)
{
await sutProvider.Sut.HandleEventAsync(eventMessage);
await sutProvider.GetDependency<IEventWriteService>().Received(1).CreateAsync(
Arg.Is(AssertHelper.AssertPropertyEqual<IEvent>(eventMessage))
);
}
}

66
test/Core.Test/AdminConsole/Services/HttpPostEventHandlerTests.cs Normal file
View File

@ -0,0 +1,66 @@
using System.Net;
using System.Net.Http.Json;
using Bit.Core.Models.Data;
using Bit.Core.Services;
using Bit.Test.Common.AutoFixture;
using Bit.Test.Common.AutoFixture.Attributes;
using Bit.Test.Common.Helpers;
using Bit.Test.Common.MockedHttpClient;
using NSubstitute;
using Xunit;
using GlobalSettings = Bit.Core.Settings.GlobalSettings;
namespace Bit.Core.Test.Services;
[SutProviderCustomize]
public class HttpPostEventHandlerTests
{
private readonly MockedHttpMessageHandler _handler;
private HttpClient _httpClient;
private const string _httpPostUrl = "http://localhost/test/event";
public HttpPostEventHandlerTests()
{
_handler = new MockedHttpMessageHandler();
_handler.Fallback
.WithStatusCode(HttpStatusCode.OK)
.WithContent(new StringContent("<html><head><title>test</title></head><body>test</body></html>"));
_httpClient = _handler.ToHttpClient();
}
public SutProvider<HttpPostEventHandler> GetSutProvider()
{
var clientFactory = Substitute.For<IHttpClientFactory>();
clientFactory.CreateClient(HttpPostEventHandler.HttpClientName).Returns(_httpClient);
var globalSettings = new GlobalSettings();
globalSettings.EventLogging.RabbitMq.HttpPostUrl = _httpPostUrl;
return new SutProvider<HttpPostEventHandler>()
.SetDependency(globalSettings)
.SetDependency(clientFactory)
.Create();
}
[Theory, BitAutoData]
public async Task HandleEventAsync_PostsEventsToUrl(EventMessage eventMessage)
{
var sutProvider = GetSutProvider();
var content = JsonContent.Create(eventMessage);
await sutProvider.Sut.HandleEventAsync(eventMessage);
sutProvider.GetDependency<IHttpClientFactory>().Received(1).CreateClient(
Arg.Is(AssertHelper.AssertPropertyEqual<string>(HttpPostEventHandler.HttpClientName))
);
Assert.Single(_handler.CapturedRequests);
var request = _handler.CapturedRequests[0];
Assert.NotNull(request);
var returned = await request.Content.ReadFromJsonAsync<EventMessage>();
Assert.Equal(HttpMethod.Post, request.Method);
Assert.Equal(_httpPostUrl, request.RequestUri.ToString());
AssertHelper.AssertPropertyEqual(eventMessage, returned, new[] { "IdempotencyId" });
}
}