rabbitmq abstracted from application
This commit is contained in:
@@ -2,5 +2,5 @@ namespace AipsCore.Application.Abstract.MessageBroking;
|
|||||||
|
|
||||||
public interface IMessagePublisher
|
public interface IMessagePublisher
|
||||||
{
|
{
|
||||||
Task PublishAsync<T>(string exchange, string routeKey, T message, CancellationToken cancellationToken = default);
|
Task PublishAsync<T>(T message, CancellationToken cancellationToken = default);
|
||||||
}
|
}
|
||||||
@@ -7,6 +7,7 @@ public static class ConfigurationEnvExtensions
|
|||||||
private const string DbConnStringKey = "DB_CONN_STRING";
|
private const string DbConnStringKey = "DB_CONN_STRING";
|
||||||
|
|
||||||
private const string RabbitMqAmqpUriKey = "RABBITMQ_AMQP_URI";
|
private const string RabbitMqAmqpUriKey = "RABBITMQ_AMQP_URI";
|
||||||
|
private const string RabbitMqExchange = "RABBITMQ_EXCHANGE";
|
||||||
|
|
||||||
private const string JwtIssuer = "JWT_ISSUER";
|
private const string JwtIssuer = "JWT_ISSUER";
|
||||||
private const string JwtAudience = "JWT_AUDIENCE";
|
private const string JwtAudience = "JWT_AUDIENCE";
|
||||||
@@ -25,6 +26,11 @@ public static class ConfigurationEnvExtensions
|
|||||||
return configuration.GetEnvForSure(RabbitMqAmqpUriKey);
|
return configuration.GetEnvForSure(RabbitMqAmqpUriKey);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public string GetEnvRabbitMqExchange()
|
||||||
|
{
|
||||||
|
return configuration.GetEnvForSure(RabbitMqExchange);
|
||||||
|
}
|
||||||
|
|
||||||
public string GetEnvJwtIssuer()
|
public string GetEnvJwtIssuer()
|
||||||
{
|
{
|
||||||
return configuration.GetEnvForSure(JwtIssuer);
|
return configuration.GetEnvForSure(JwtIssuer);
|
||||||
|
|||||||
@@ -2,6 +2,8 @@ using System.Text;
|
|||||||
using System.Text.Json;
|
using System.Text.Json;
|
||||||
using AipsCore.Application.Abstract;
|
using AipsCore.Application.Abstract;
|
||||||
using AipsCore.Application.Abstract.MessageBroking;
|
using AipsCore.Application.Abstract.MessageBroking;
|
||||||
|
using AipsCore.Infrastructure.DI.Configuration;
|
||||||
|
using Microsoft.Extensions.Configuration;
|
||||||
using RabbitMQ.Client;
|
using RabbitMQ.Client;
|
||||||
|
|
||||||
namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
|
namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
|
||||||
@@ -9,20 +11,26 @@ namespace AipsCore.Infrastructure.MessageBroking.RabbitMQ;
|
|||||||
public class RabbitMqPublisher : IMessagePublisher
|
public class RabbitMqPublisher : IMessagePublisher
|
||||||
{
|
{
|
||||||
private readonly IRabbitMqConnection _connection;
|
private readonly IRabbitMqConnection _connection;
|
||||||
|
private readonly IConfiguration _configuration;
|
||||||
|
|
||||||
public RabbitMqPublisher(IRabbitMqConnection connection)
|
public RabbitMqPublisher(IRabbitMqConnection connection, IConfiguration configuration)
|
||||||
{
|
{
|
||||||
_connection = connection;
|
_connection = connection;
|
||||||
|
_configuration = configuration;
|
||||||
}
|
}
|
||||||
|
|
||||||
public async Task PublishAsync<T>(string exchange, string routeKey, T message, CancellationToken cancellationToken = default)
|
public async Task PublishAsync<T>(T message, CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
var channel = await _connection.CreateChannelAsync(cancellationToken);
|
var channel = await _connection.CreateChannelAsync(cancellationToken);
|
||||||
|
|
||||||
await channel.ExchangeDeclareAsync(exchange, ExchangeType.Topic, durable: true, cancellationToken: cancellationToken);
|
await channel.ExchangeDeclareAsync(GetExchange(), ExchangeType.Topic, durable: true, cancellationToken: cancellationToken);
|
||||||
|
|
||||||
var bytes = Serialize(message);
|
var bytes = Serialize(message);
|
||||||
await channel.BasicPublishAsync(exchange, routeKey, bytes, cancellationToken);
|
await channel.BasicPublishAsync(
|
||||||
|
GetExchange(),
|
||||||
|
typeof(T).Name,
|
||||||
|
bytes,
|
||||||
|
cancellationToken);
|
||||||
|
|
||||||
await channel.CloseAsync(cancellationToken);
|
await channel.CloseAsync(cancellationToken);
|
||||||
}
|
}
|
||||||
@@ -31,4 +39,6 @@ public class RabbitMqPublisher : IMessagePublisher
|
|||||||
{
|
{
|
||||||
return JsonSerializer.SerializeToUtf8Bytes(message);
|
return JsonSerializer.SerializeToUtf8Bytes(message);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private string GetExchange() => _configuration.GetEnvRabbitMqExchange();
|
||||||
}
|
}
|
||||||
@@ -40,6 +40,6 @@ public class UserController : ControllerBase
|
|||||||
[HttpPost("test")]
|
[HttpPost("test")]
|
||||||
public async Task Test(IMessagePublisher publisher)
|
public async Task Test(IMessagePublisher publisher)
|
||||||
{
|
{
|
||||||
await publisher.PublishAsync("test", "test.test", "Test poruka");
|
await publisher.PublishAsync("Test poruka");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Reference in New Issue
Block a user