91 lines
3.0 KiB
C#
91 lines
3.0 KiB
C#
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Options;
|
|
using RabbitMQ.Client;
|
|
using RabbitMQ.Client.Exceptions;
|
|
|
|
public sealed class RabbitMqTopologyProvisioner : IHostedService
|
|
{
|
|
public const string Exchange = "clinical.notifications.exchange";
|
|
public const string PagingKey = "alerts.paging";
|
|
public const string EscalKey = "alerts.escalation";
|
|
|
|
private readonly RabbitMqOptions _opts;
|
|
private readonly IHostEnvironment _env;
|
|
private readonly ILogger<RabbitMqTopologyProvisioner> _logger;
|
|
|
|
public RabbitMqTopologyProvisioner(
|
|
IOptions<RabbitMqOptions> opts,
|
|
IHostEnvironment env,
|
|
ILogger<RabbitMqTopologyProvisioner> logger)
|
|
{
|
|
_opts = opts.Value;
|
|
_env = env;
|
|
_logger = logger;
|
|
}
|
|
|
|
public Task StartAsync(CancellationToken ct)
|
|
{
|
|
var factory = BuildFactory();
|
|
using var connection = factory.CreateConnection("gateway-topology");
|
|
using var channel = connection.CreateModel();
|
|
|
|
if (_env.IsDevelopment() || _env.EnvironmentName == "Testing")
|
|
{
|
|
using var cleanup = connection.CreateModel();
|
|
try
|
|
{
|
|
cleanup.QueueDelete("alerts.paging.dlq", ifUnused: false, ifEmpty: false);
|
|
}
|
|
catch (OperationInterruptedException ex)
|
|
{
|
|
_logger.LogDebug(ex, "DLQ delete skipped — queue may not exist yet");
|
|
}
|
|
}
|
|
|
|
channel.ExchangeDeclare(Exchange, ExchangeType.Direct, durable: true);
|
|
|
|
channel.QueueDeclare(
|
|
queue: "alerts.paging.queue",
|
|
durable: true,
|
|
exclusive: false,
|
|
autoDelete: false,
|
|
arguments: new Dictionary<string, object>
|
|
{
|
|
["x-dead-letter-exchange"] = "",
|
|
["x-dead-letter-routing-key"] = "alerts.paging.dlq",
|
|
});
|
|
channel.QueueBind("alerts.paging.queue", Exchange, PagingKey);
|
|
|
|
channel.QueueDeclare(
|
|
queue: "alerts.paging.dlq",
|
|
durable: true,
|
|
exclusive: false,
|
|
autoDelete: false,
|
|
arguments: new Dictionary<string, object>
|
|
{
|
|
["x-message-ttl"] = _opts.PagingAckTimeoutMs,
|
|
["x-dead-letter-exchange"] = Exchange,
|
|
["x-dead-letter-routing-key"] = EscalKey,
|
|
});
|
|
|
|
channel.QueueDeclare(
|
|
queue: "alerts.escalation.queue",
|
|
durable: true,
|
|
exclusive: false,
|
|
autoDelete: false,
|
|
arguments: null);
|
|
channel.QueueBind("alerts.escalation.queue", Exchange, EscalKey);
|
|
|
|
_logger.LogInformation(
|
|
"Gateway RabbitMQ topology provisioned. Exchange={Exchange} PagingDlqTtlMs={Ttl}",
|
|
Exchange, _opts.PagingAckTimeoutMs);
|
|
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public Task StopAsync(CancellationToken ct) => Task.CompletedTask;
|
|
|
|
public ConnectionFactory BuildFactory() =>
|
|
RabbitMqConnectionFactory.Create(_opts, dispatchConsumersAsync: true);
|
|
}
|