RabbitMqBus.cs
8,265 bytes
| 1 | using System.Collections.Concurrent; |
|---|---|
| 2 | using System.Text.Json; |
| 3 | using Microsoft.Extensions.Logging; |
| 4 | using Microsoft.Extensions.Options; |
| 5 | using RabbitMQ.Client; |
| 6 | using RabbitMQ.Client.Events; |
| 7 | |
| 8 | namespace SplitApp.Shared.Messaging.Internal; |
| 9 | |
| 10 | /// <summary> |
| 11 | /// Singleton IMessageBus implementation. Owns: |
| 12 | /// - one publish channel (events + RPC outbound), |
| 13 | /// - one reply consumer channel + exclusive server-named auto-delete reply queue, |
| 14 | /// - a TCS map keyed by correlation ID for in-flight RPC requests. |
| 15 | /// |
| 16 | /// On connection/channel shutdown, all pending RPC TCSs are faulted with |
| 17 | /// <see cref="MessageBusUnavailableException"/> so callers fail fast. |
| 18 | /// </summary> |
| 19 | public class RabbitMqBus : IMessageBus, IAsyncDisposable |
| 20 | { |
| 21 | private readonly RabbitMqConnectionProvider _connections; |
| 22 | private readonly MessagingOptions _options; |
| 23 | private readonly ILogger<RabbitMqBus> _logger; |
| 24 | private readonly SemaphoreSlim _initGate = new(1, 1); |
| 25 | private readonly ConcurrentDictionary<string, TaskCompletionSource<ReadOnlyMemory<byte>>> _pending = new(); |
| 26 | |
| 27 | private IChannel? _publishChannel; |
| 28 | private IChannel? _replyChannel; |
| 29 | private string? _replyQueueName; |
| 30 | private bool _topologyDeclared; |
| 31 | |
| 32 | public RabbitMqBus( |
| 33 | RabbitMqConnectionProvider connections, |
| 34 | IOptions<MessagingOptions> options, |
| 35 | ILogger<RabbitMqBus> logger) |
| 36 | { |
| 37 | _connections = connections; |
| 38 | _options = options.Value; |
| 39 | _logger = logger; |
| 40 | } |
| 41 | |
| 42 | public async Task PublishEventAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default) |
| 43 | where TEvent : IIntegrationEvent |
| 44 | { |
| 45 | await EnsureInitializedAsync(cancellationToken); |
| 46 | |
| 47 | var typeName = typeof(TEvent).Name; |
| 48 | var body = JsonSerializer.SerializeToUtf8Bytes(@event); |
| 49 | var props = new BasicProperties |
| 50 | { |
| 51 | Type = typeName, |
| 52 | MessageId = Guid.NewGuid().ToString(), |
| 53 | ContentType = "application/json", |
| 54 | DeliveryMode = DeliveryModes.Persistent, |
| 55 | }; |
| 56 | |
| 57 | await _publishChannel!.BasicPublishAsync( |
| 58 | exchange: RabbitMqTopology.EventsExchange, |
| 59 | routingKey: typeName, |
| 60 | mandatory: false, |
| 61 | basicProperties: props, |
| 62 | body: body, |
| 63 | cancellationToken: cancellationToken); |
| 64 | |
| 65 | _logger.LogDebug("Published event {Type} ({MessageId})", typeName, props.MessageId); |
| 66 | } |
| 67 | |
| 68 | public async Task<TResponse> RequestAsync<TRequest, TResponse>( |
| 69 | TRequest request, |
| 70 | TimeSpan timeout, |
| 71 | CancellationToken cancellationToken = default) |
| 72 | where TRequest : IIntegrationRequest<TResponse> |
| 73 | { |
| 74 | await EnsureInitializedAsync(cancellationToken); |
| 75 | |
| 76 | var typeName = typeof(TRequest).Name; |
| 77 | var correlationId = Guid.NewGuid().ToString(); |
| 78 | var tcs = new TaskCompletionSource<ReadOnlyMemory<byte>>(TaskCreationOptions.RunContinuationsAsynchronously); |
| 79 | _pending[correlationId] = tcs; |
| 80 | |
| 81 | try |
| 82 | { |
| 83 | var body = JsonSerializer.SerializeToUtf8Bytes(request); |
| 84 | var props = new BasicProperties |
| 85 | { |
| 86 | Type = typeName, |
| 87 | CorrelationId = correlationId, |
| 88 | ReplyTo = _replyQueueName, |
| 89 | ContentType = "application/json", |
| 90 | }; |
| 91 | |
| 92 | await _publishChannel!.BasicPublishAsync( |
| 93 | exchange: RabbitMqTopology.RequestsExchange, |
| 94 | routingKey: typeName, |
| 95 | mandatory: false, |
| 96 | basicProperties: props, |
| 97 | body: body, |
| 98 | cancellationToken: cancellationToken); |
| 99 | |
| 100 | _logger.LogDebug("Sent request {Type} ({CorrelationId})", typeName, correlationId); |
| 101 | |
| 102 | using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); |
| 103 | timeoutCts.CancelAfter(timeout); |
| 104 | |
| 105 | ReadOnlyMemory<byte> replyBody; |
| 106 | try |
| 107 | { |
| 108 | replyBody = await tcs.Task.WaitAsync(timeoutCts.Token); |
| 109 | } |
| 110 | catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) |
| 111 | { |
| 112 | throw new MessageBusTimeoutException( |
| 113 | $"RPC request {typeName} ({correlationId}) timed out after {timeout.TotalSeconds:0.#}s"); |
| 114 | } |
| 115 | |
| 116 | var response = JsonSerializer.Deserialize<TResponse>(replyBody.Span); |
| 117 | return response!; |
| 118 | } |
| 119 | finally |
| 120 | { |
| 121 | _pending.TryRemove(correlationId, out _); |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | private async Task EnsureInitializedAsync(CancellationToken ct) |
| 126 | { |
| 127 | if (_publishChannel is { IsOpen: true } && _replyChannel is { IsOpen: true }) return; |
| 128 | |
| 129 | await _initGate.WaitAsync(ct); |
| 130 | try |
| 131 | { |
| 132 | if (_publishChannel is { IsOpen: true } && _replyChannel is { IsOpen: true }) return; |
| 133 | |
| 134 | var conn = await _connections.GetConnectionAsync(ct); |
| 135 | _publishChannel ??= await conn.CreateChannelAsync(cancellationToken: ct); |
| 136 | _replyChannel ??= await conn.CreateChannelAsync(cancellationToken: ct); |
| 137 | |
| 138 | if (!_topologyDeclared) |
| 139 | { |
| 140 | await _publishChannel.ExchangeDeclareAsync( |
| 141 | RabbitMqTopology.EventsExchange, |
| 142 | RabbitMqTopology.EventsExchangeType, |
| 143 | durable: true, |
| 144 | autoDelete: false, |
| 145 | cancellationToken: ct); |
| 146 | |
| 147 | await _publishChannel.ExchangeDeclareAsync( |
| 148 | RabbitMqTopology.RequestsExchange, |
| 149 | RabbitMqTopology.RequestsExchangeType, |
| 150 | durable: true, |
| 151 | autoDelete: false, |
| 152 | cancellationToken: ct); |
| 153 | |
| 154 | _topologyDeclared = true; |
| 155 | } |
| 156 | |
| 157 | if (_replyQueueName is null) |
| 158 | { |
| 159 | var declareOk = await _replyChannel.QueueDeclareAsync( |
| 160 | queue: "", |
| 161 | durable: false, |
| 162 | exclusive: true, |
| 163 | autoDelete: true, |
| 164 | cancellationToken: ct); |
| 165 | _replyQueueName = declareOk.QueueName; |
| 166 | |
| 167 | var consumer = new AsyncEventingBasicConsumer(_replyChannel); |
| 168 | consumer.ReceivedAsync += OnReplyReceivedAsync; |
| 169 | await _replyChannel.BasicConsumeAsync( |
| 170 | queue: _replyQueueName, |
| 171 | autoAck: false, |
| 172 | consumer: consumer, |
| 173 | cancellationToken: ct); |
| 174 | |
| 175 | _logger.LogInformation("RPC reply queue ready: {Queue}", _replyQueueName); |
| 176 | } |
| 177 | |
| 178 | conn.ConnectionShutdownAsync += OnConnectionShutdownAsync; |
| 179 | } |
| 180 | finally |
| 181 | { |
| 182 | _initGate.Release(); |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | private Task OnReplyReceivedAsync(object sender, BasicDeliverEventArgs ea) |
| 187 | { |
| 188 | var corr = ea.BasicProperties.CorrelationId; |
| 189 | if (!string.IsNullOrEmpty(corr) && _pending.TryRemove(corr, out var tcs)) |
| 190 | { |
| 191 | tcs.TrySetResult(ea.Body); |
| 192 | } |
| 193 | else |
| 194 | { |
| 195 | _logger.LogWarning("Reply with unknown correlation id {CorrelationId} dropped", corr); |
| 196 | } |
| 197 | |
| 198 | return _replyChannel!.BasicAckAsync(ea.DeliveryTag, multiple: false).AsTask(); |
| 199 | } |
| 200 | |
| 201 | private Task OnConnectionShutdownAsync(object? sender, ShutdownEventArgs ea) |
| 202 | { |
| 203 | _logger.LogWarning("RabbitMQ connection shutdown: {Reason}", ea.ReplyText); |
| 204 | var ex = new MessageBusUnavailableException($"RabbitMQ connection lost: {ea.ReplyText}"); |
| 205 | foreach (var kvp in _pending) |
| 206 | { |
| 207 | if (_pending.TryRemove(kvp.Key, out var tcs)) |
| 208 | { |
| 209 | tcs.TrySetException(ex); |
| 210 | } |
| 211 | } |
| 212 | return Task.CompletedTask; |
| 213 | } |
| 214 | |
| 215 | public async ValueTask DisposeAsync() |
| 216 | { |
| 217 | try |
| 218 | { |
| 219 | if (_publishChannel is not null) { await _publishChannel.CloseAsync(); await _publishChannel.DisposeAsync(); } |
| 220 | if (_replyChannel is not null) { await _replyChannel.CloseAsync(); await _replyChannel.DisposeAsync(); } |
| 221 | } |
| 222 | catch (Exception ex) |
| 223 | { |
| 224 | _logger.LogWarning(ex, "Error closing RabbitMQ bus channels"); |
| 225 | } |
| 226 | _initGate.Dispose(); |
| 227 | } |
| 228 | } |
| 229 | |