| | | 1 | | using Microsoft.Extensions.Options; |
| | | 2 | | using System.Diagnostics; |
| | | 3 | | using System.Text; |
| | | 4 | | using System.Text.Json; |
| | | 5 | | |
| | | 6 | | namespace AsyncResponse.Transports.RabbitMQ; |
| | | 7 | | |
| | | 8 | | /// <summary> |
| | | 9 | | /// Publishes <see cref="WorkerJobEnvelope"/> messages to a RabbitMQ exchange. |
| | | 10 | | /// </summary> |
| | | 11 | | /// <remarks> |
| | | 12 | | /// The publish channel is created lazily and re-created on demand: a transient broker outage when the |
| | | 13 | | /// first job is published no longer permanently breaks the transport (a faulted connect attempt is not |
| | | 14 | | /// cached). The channel is opened with publisher confirmations so <see cref="PublishAsync"/> only completes |
| | | 15 | | /// once the broker has accepted the message. A single channel is shared across concurrent publishers; |
| | | 16 | | /// RabbitMQ.Client v7 tracks each in-flight confirmation independently, so concurrent publishing is safe. |
| | | 17 | | /// </remarks> |
| | | 18 | | public sealed class RabbitMqWorkerTransport : IWorkerTransport, IAsyncDisposable |
| | | 19 | | { |
| | | 20 | | private readonly RabbitMqAsyncResponseOptions _options; |
| | | 21 | | private readonly IRabbitMqConnectionFactory _connectionFactory; |
| | 3 | 22 | | private readonly SemaphoreSlim _connectionGate = new(1, 1); |
| | | 23 | | private IRabbitMqConnection? _connection; |
| | | 24 | | private IRabbitMqChannel? _channel; |
| | | 25 | | private int _disposeGate; |
| | | 26 | | private bool _disposed; |
| | | 27 | | |
| | | 28 | | /// <summary>Runs the RabbitMqWorkerTransport operation.</summary> |
| | | 29 | | public RabbitMqWorkerTransport(IOptions<RabbitMqAsyncResponseOptions> options) |
| | 3 | 30 | | : this(options, new RabbitMqConnectionFactoryAdapter(options.Value)) |
| | | 31 | | { |
| | 3 | 32 | | } |
| | | 33 | | |
| | 3 | 34 | | internal RabbitMqWorkerTransport( |
| | 3 | 35 | | IOptions<RabbitMqAsyncResponseOptions> options, |
| | 3 | 36 | | IRabbitMqConnectionFactory connectionFactory) |
| | | 37 | | { |
| | 3 | 38 | | _options = options.Value; |
| | 3 | 39 | | ValidatePublishOptions(_options); |
| | 3 | 40 | | _connectionFactory = connectionFactory; |
| | 3 | 41 | | } |
| | | 42 | | |
| | | 43 | | private static void ValidatePublishOptions(RabbitMqAsyncResponseOptions options) |
| | | 44 | | { |
| | 3 | 45 | | _ = RabbitMqOptionsValidator.Required(options.WorkerExchange, nameof(options.WorkerExchange)); |
| | 3 | 46 | | _ = RabbitMqOptionsValidator.Required(options.WorkerQueue, nameof(options.WorkerQueue)); |
| | 3 | 47 | | _ = RabbitMqOptionsValidator.Required(options.WorkerRoutingKey, nameof(options.WorkerRoutingKey)); |
| | 3 | 48 | | RabbitMqOptionsValidator.Positive(options.ShutdownTimeout, nameof(options.ShutdownTimeout)); |
| | 3 | 49 | | } |
| | | 50 | | |
| | | 51 | | private async Task<IRabbitMqChannel> GetChannelAsync(CancellationToken cancellationToken) |
| | | 52 | | { |
| | 3 | 53 | | var channel = Volatile.Read(ref _channel); |
| | 3 | 54 | | if (channel is not null) |
| | 3 | 55 | | return channel; |
| | | 56 | | |
| | 3 | 57 | | await _connectionGate.WaitAsync(cancellationToken).ConfigureAwait(false); |
| | | 58 | | try |
| | | 59 | | { |
| | 3 | 60 | | ObjectDisposedException.ThrowIf(_disposed, this); |
| | 3 | 61 | | if (_channel is not null) |
| | 1 | 62 | | return _channel; |
| | | 63 | | |
| | | 64 | | // ??= only assigns when the await succeeds, so a failed connect leaves _connection null and |
| | | 65 | | // the next publish retries. A successful connection is reused even if channel/topology setup fails. |
| | 3 | 66 | | _connection ??= await _connectionFactory.CreateConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 67 | | var created = await _connection.CreateChannelAsync(publisherConfirmations: true, cancellationToken).Configur |
| | 3 | 68 | | await RabbitMqTopology.EnsureWorkerAsync(created, _options, cancellationToken).ConfigureAwait(false); |
| | | 69 | | |
| | | 70 | | // Publish the channel only once it is fully initialized; if anything above threw, _channel stays |
| | | 71 | | // null so a later publish recreates it instead of awaiting a permanently faulted task. |
| | 3 | 72 | | _channel = created; |
| | 3 | 73 | | return created; |
| | | 74 | | } |
| | | 75 | | finally |
| | | 76 | | { |
| | 3 | 77 | | _connectionGate.Release(); |
| | | 78 | | } |
| | 3 | 79 | | } |
| | | 80 | | |
| | | 81 | | /// <summary>Publishes the supplied message.</summary> |
| | | 82 | | public async Task PublishAsync(WorkerJobEnvelope job, CancellationToken cancellationToken = default) |
| | | 83 | | { |
| | 3 | 84 | | ArgumentNullException.ThrowIfNull(job); |
| | | 85 | | |
| | 3 | 86 | | using var activity = AsyncResponseDiagnostics.StartActivity( |
| | 3 | 87 | | "asyncresponse.worker.publish", |
| | 3 | 88 | | ActivityKind.Producer, |
| | 3 | 89 | | job.CorrelationId); |
| | 3 | 90 | | activity?.SetTag("asyncresponse.transport", "rabbitmq"); |
| | 3 | 91 | | activity?.SetTag("messaging.system", "rabbitmq"); |
| | 3 | 92 | | activity?.SetTag("messaging.destination.name", _options.WorkerExchange); |
| | 3 | 93 | | activity?.SetTag("messaging.rabbitmq.routing_key", _options.WorkerRoutingKey); |
| | 3 | 94 | | AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget); |
| | 3 | 95 | | AsyncResponseDiagnostics.SetWorker(activity, job.Call); |
| | | 96 | | |
| | | 97 | | try |
| | | 98 | | { |
| | 3 | 99 | | var payload = Encoding.UTF8.GetBytes(AsyncResponseJson.Serialize(job)); |
| | 3 | 100 | | var properties = RabbitMqTopology.CreatePersistentJsonProperties(job.CorrelationId, _options.CorrelationIdHe |
| | 3 | 101 | | var channel = await GetChannelAsync(cancellationToken).ConfigureAwait(false); |
| | 3 | 102 | | await channel.BasicPublishAsync( |
| | 3 | 103 | | _options.WorkerExchange, |
| | 3 | 104 | | _options.WorkerRoutingKey, |
| | 3 | 105 | | properties, |
| | 3 | 106 | | payload, |
| | 3 | 107 | | cancellationToken).ConfigureAwait(false); |
| | 3 | 108 | | activity?.SetTag("messaging.message.id", properties.MessageId); |
| | 3 | 109 | | } |
| | 2 | 110 | | catch (Exception ex) |
| | | 111 | | { |
| | 2 | 112 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 3 | 113 | | throw; |
| | | 114 | | } |
| | 3 | 115 | | } |
| | | 116 | | |
| | | 117 | | /// <summary>Releases resources held by this instance.</summary> |
| | | 118 | | public async ValueTask DisposeAsync() |
| | | 119 | | { |
| | 3 | 120 | | if (Interlocked.Exchange(ref _disposeGate, 1) != 0) |
| | 3 | 121 | | return; |
| | | 122 | | |
| | 3 | 123 | | await _connectionGate.WaitAsync().ConfigureAwait(false); |
| | | 124 | | try |
| | | 125 | | { |
| | 3 | 126 | | _disposed = true; |
| | | 127 | | |
| | 3 | 128 | | if (_channel is not null) |
| | | 129 | | { |
| | 3 | 130 | | using var cts = new CancellationTokenSource(_options.ShutdownTimeout); |
| | | 131 | | try |
| | | 132 | | { |
| | 3 | 133 | | await _channel.CloseAsync(cts.Token).ConfigureAwait(false); |
| | 3 | 134 | | } |
| | 3 | 135 | | catch |
| | | 136 | | { |
| | | 137 | | // Best effort: the channel may already be closed by broker-side shutdown. |
| | 3 | 138 | | } |
| | | 139 | | |
| | 3 | 140 | | await _channel.DisposeAsync().ConfigureAwait(false); |
| | 3 | 141 | | } |
| | | 142 | | |
| | 3 | 143 | | if (_connection is not null) |
| | | 144 | | { |
| | 3 | 145 | | using var cts = new CancellationTokenSource(_options.ShutdownTimeout); |
| | | 146 | | try |
| | | 147 | | { |
| | 3 | 148 | | await _connection.CloseAsync(_options.ShutdownTimeout, cts.Token).ConfigureAwait(false); |
| | 3 | 149 | | } |
| | 3 | 150 | | catch |
| | | 151 | | { |
| | | 152 | | // Best effort. |
| | 3 | 153 | | } |
| | | 154 | | |
| | 3 | 155 | | await _connection.DisposeAsync().ConfigureAwait(false); |
| | 3 | 156 | | } |
| | 3 | 157 | | } |
| | | 158 | | finally |
| | | 159 | | { |
| | 3 | 160 | | _connectionGate.Release(); |
| | 3 | 161 | | _connectionGate.Dispose(); |
| | | 162 | | } |
| | 3 | 163 | | } |
| | | 164 | | } |