implemented rabbitmq message broker support (publisher)

This commit is contained in:
2026-02-13 23:40:54 +01:00
parent ff2daa64b5
commit e2db689888
6 changed files with 100 additions and 0 deletions

View File

@@ -0,0 +1,8 @@
using RabbitMQ.Client;
namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
public interface IRabbitMqConnection
{
Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default);
}

View File

@@ -0,0 +1,43 @@
using AipsCore.Infrastructure.DI.Configuration;
using Microsoft.Extensions.Configuration;
using RabbitMQ.Client;
namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
public class RabbitMqConnection : IRabbitMqConnection
{
private readonly IConfiguration _configuration;
private IConnection? _connection;
public RabbitMqConnection(IConfiguration configuration)
{
_configuration = configuration;
}
public async Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)
{
if (_connection is null)
{
throw new InvalidOperationException($"RabbitMQ connection not created for {_configuration.GetEnvRabbitMqAmqpUri()}");
}
return await _connection.CreateChannelAsync(null, cancellationToken);
}
public async Task CreateConnectionAsync()
{
var factory = CreateConnectionFactory();
_connection = await factory.CreateConnectionAsync();
}
private IConnectionFactory CreateConnectionFactory()
{
var factory = new ConnectionFactory
{
Uri = new Uri(_configuration.GetEnvRabbitMqAmqpUri())
};
return factory;
}
}

View File

@@ -0,0 +1,33 @@
using System.Text;
using System.Text.Json;
using AipsCore.Application.Abstract;
using RabbitMQ.Client;
namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
public class RabbitMqPublisher : IMessagePublisher
{
private readonly IRabbitMqConnection _connection;
public RabbitMqPublisher(IRabbitMqConnection connection)
{
_connection = connection;
}
public async Task PublishAsync<T>(string exchange, string routeKey, T message, CancellationToken cancellationToken = default)
{
var channel = await _connection.CreateChannelAsync(cancellationToken);
await channel.ExchangeDeclareAsync(exchange, ExchangeType.Topic, durable: true, cancellationToken: cancellationToken);
var bytes = Serialize(message);
await channel.BasicPublishAsync(exchange, routeKey, bytes, cancellationToken);
await channel.CloseAsync(cancellationToken);
}
private byte[] Serialize<T>(T message)
{
return JsonSerializer.SerializeToUtf8Bytes(message);
}
}