bitwarden/server
1
0
Fork 0
mirror of https://github.com/bitwarden/server synced 2025-12-22 11:13:27 +00:00
Code Issues Releases Wiki Activity

[PM-17562] Initial POC of Distributed Events (#5323)

Browse Source 
* Initial POC of Distributed Events

* Apply suggestions from code review

Co-authored-by: Justin Baur <19896123+justindbaur@users.noreply.github.com>

* Clean up files to support accepted changes. Address PR Feedback

* Removed unneeded using to fix lint warning

* Moved config into a common EventLogging top-level item. Fixed issues from PR review

* Optimized per suggestion from justinbaur

Co-authored-by: Justin Baur <19896123+justindbaur@users.noreply.github.com>

* Updated to add IAsyncDisposable as suggested in PR review

* Updated with suggestion to use KeyedSingleton for the IEventWriteService

* Changed key case to lowercase

---------

Co-authored-by: Justin Baur <19896123+justindbaur@users.noreply.github.com>
This commit is contained in:
Brant DeBow
2025-01-30 11:07:02 -06:00
committed by GitHub
parent 443a147433
commit 5efd68cf51
10 changed files with 311 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
src/Core/AdminConsole/Services/Implementations/RabbitMqEventWriteService.cs Normal file
View File

@@ -0,0 +1,65 @@
using System.Text.Json;
using Bit.Core.Models.Data;
using Bit.Core.Settings;
using RabbitMQ.Client;
namespace Bit.Core.Services;
public class RabbitMqEventWriteService : IEventWriteService, IAsyncDisposable
{
private readonly ConnectionFactory _factory;
private readonly Lazy<Task<IConnection>> _lazyConnection;
private readonly string _exchangeName;
public RabbitMqEventWriteService(GlobalSettings globalSettings)
{
_factory = new ConnectionFactory
{
HostName = globalSettings.EventLogging.RabbitMq.HostName,
UserName = globalSettings.EventLogging.RabbitMq.Username,
Password = globalSettings.EventLogging.RabbitMq.Password
};
_exchangeName = globalSettings.EventLogging.RabbitMq.ExchangeName;
_lazyConnection = new Lazy<Task<IConnection>>(CreateConnectionAsync);
}
public async Task CreateAsync(IEvent e)
{
var connection = await _lazyConnection.Value;
using var channel = await connection.CreateChannelAsync();
await channel.ExchangeDeclareAsync(exchange: _exchangeName, type: ExchangeType.Fanout, durable: true);
var body = JsonSerializer.SerializeToUtf8Bytes(e);
await channel.BasicPublishAsync(exchange: _exchangeName, routingKey: string.Empty, body: body);
}
public async Task CreateManyAsync(IEnumerable<IEvent> events)
{
var connection = await _lazyConnection.Value;
using var channel = await connection.CreateChannelAsync();
await channel.ExchangeDeclareAsync(exchange: _exchangeName, type: ExchangeType.Fanout, durable: true);
foreach (var e in events)
{
var body = JsonSerializer.SerializeToUtf8Bytes(e);
await channel.BasicPublishAsync(exchange: _exchangeName, routingKey: string.Empty, body: body);
}
}
public async ValueTask DisposeAsync()
{
if (_lazyConnection.IsValueCreated)
{
var connection = await _lazyConnection.Value;
await connection.DisposeAsync();
}
}
private async Task<IConnection> CreateConnectionAsync()
{
return await _factory.CreateConnectionAsync();
}
}