| | | 1 | | using RabbitMQ.Client; |
| | | 2 | | |
| | | 3 | | namespace AsyncResponse.Transports.RabbitMQ; |
| | | 4 | | |
| | | 5 | | internal static class RabbitMqTopology |
| | | 6 | | { |
| | | 7 | | private const string DirectExchange = "direct"; |
| | | 8 | | |
| | | 9 | | /// <summary>Ensures the required resource exists.</summary> |
| | | 10 | | public static Task EnsureWorkerAsync( |
| | | 11 | | IRabbitMqChannel channel, |
| | | 12 | | RabbitMqAsyncResponseOptions options, |
| | | 13 | | CancellationToken cancellationToken = default) |
| | | 14 | | { |
| | 439 | 15 | | if (!options.DeclareTopology) |
| | 2 | 16 | | return Task.CompletedTask; |
| | | 17 | | |
| | 437 | 18 | | var exchange = RabbitMqOptionsValidator.Required(options.WorkerExchange, nameof(options.WorkerExchange)); |
| | 437 | 19 | | var queue = RabbitMqOptionsValidator.Required(options.WorkerQueue, nameof(options.WorkerQueue)); |
| | 437 | 20 | | var routingKey = RabbitMqOptionsValidator.Required(options.WorkerRoutingKey, nameof(options.WorkerRoutingKey)); |
| | | 21 | | |
| | 437 | 22 | | return DeclareQueueTopologyAsync(channel, options, exchange, queue, routingKey, cancellationToken); |
| | | 23 | | } |
| | | 24 | | |
| | | 25 | | /// <summary>Ensures the required resource exists.</summary> |
| | | 26 | | public static Task EnsureResponseAsync( |
| | | 27 | | IRabbitMqChannel channel, |
| | | 28 | | RabbitMqAsyncResponseOptions options, |
| | | 29 | | CancellationToken cancellationToken = default) |
| | | 30 | | { |
| | 200 | 31 | | if (!options.DeclareTopology) |
| | 2 | 32 | | return Task.CompletedTask; |
| | | 33 | | |
| | 198 | 34 | | var exchange = RabbitMqOptionsValidator.Required(options.ResponseExchange, nameof(options.ResponseExchange)); |
| | 198 | 35 | | var queue = RabbitMqOptionsValidator.Required(options.ResponseQueue, nameof(options.ResponseQueue)); |
| | 198 | 36 | | var routingKey = RabbitMqOptionsValidator.Required(options.ResponseRoutingKey, nameof(options.ResponseRoutingKey |
| | | 37 | | |
| | 198 | 38 | | return DeclareQueueTopologyAsync(channel, options, exchange, queue, routingKey, cancellationToken); |
| | | 39 | | } |
| | | 40 | | |
| | | 41 | | private static async Task DeclareQueueTopologyAsync( |
| | | 42 | | IRabbitMqChannel channel, |
| | | 43 | | RabbitMqAsyncResponseOptions options, |
| | | 44 | | string exchange, |
| | | 45 | | string queue, |
| | | 46 | | string routingKey, |
| | | 47 | | CancellationToken cancellationToken) |
| | | 48 | | { |
| | 635 | 49 | | await channel.ExchangeDeclareAsync(exchange, DirectExchange, durable: true, autoDelete: false, cancellationToken |
| | | 50 | | |
| | 633 | 51 | | var queueArguments = await EnsureDeadLetterTopologyAsync(channel, options, routingKey, cancellationToken).Config |
| | | 52 | | |
| | | 53 | | // The park queue is declared on its own, never bound: it is reached only through the default |
| | | 54 | | // exchange (routing key = queue name) by the park-at-cap publish. Bound to the dead-letter |
| | | 55 | | // exchange — as DeadLetterQueue is — it would also collect one copy per retry hop of a |
| | | 56 | | // TTL-retry cycle, including the hops of messages that later succeed. Declared with or |
| | | 57 | | // without a dead-letter exchange: a broker policy can supply that exchange, and a park |
| | | 58 | | // destination nobody declared turns every park into a failed publish. |
| | 633 | 59 | | if (!string.IsNullOrWhiteSpace(options.ParkQueue)) |
| | 0 | 60 | | await channel.QueueDeclareAsync(options.ParkQueue, durable: true, exclusive: false, autoDelete: false, argum |
| | | 61 | | |
| | 633 | 62 | | await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false, queueArguments, cance |
| | 633 | 63 | | await channel.QueueBindAsync(queue, exchange, routingKey, cancellationToken).ConfigureAwait(false); |
| | 633 | 64 | | } |
| | | 65 | | |
| | | 66 | | /// <summary> |
| | | 67 | | /// Declares the dead-letter exchange/queue/binding when configured and returns the arguments to apply to the |
| | | 68 | | /// source queue (<c>x-dead-letter-exchange</c> and optionally <c>x-dead-letter-routing-key</c>), or null when |
| | | 69 | | /// no dead-letter exchange is configured. |
| | | 70 | | /// </summary> |
| | | 71 | | private static async Task<IDictionary<string, object?>?> EnsureDeadLetterTopologyAsync( |
| | | 72 | | IRabbitMqChannel channel, |
| | | 73 | | RabbitMqAsyncResponseOptions options, |
| | | 74 | | string sourceRoutingKey, |
| | | 75 | | CancellationToken cancellationToken) |
| | | 76 | | { |
| | 633 | 77 | | if (string.IsNullOrWhiteSpace(options.DeadLetterExchange)) |
| | 619 | 78 | | return null; |
| | | 79 | | |
| | 14 | 80 | | var deadLetterExchange = options.DeadLetterExchange; |
| | 14 | 81 | | await channel.ExchangeDeclareAsync(deadLetterExchange, DirectExchange, durable: true, autoDelete: false, cancell |
| | | 82 | | |
| | 14 | 83 | | if (!string.IsNullOrWhiteSpace(options.DeadLetterQueue)) |
| | | 84 | | { |
| | 10 | 85 | | var bindingRoutingKey = string.IsNullOrWhiteSpace(options.DeadLetterRoutingKey) |
| | 10 | 86 | | ? sourceRoutingKey |
| | 10 | 87 | | : options.DeadLetterRoutingKey; |
| | | 88 | | |
| | 10 | 89 | | await channel.QueueDeclareAsync(options.DeadLetterQueue, durable: true, exclusive: false, autoDelete: false, |
| | 10 | 90 | | await channel.QueueBindAsync(options.DeadLetterQueue, deadLetterExchange, bindingRoutingKey, cancellationTok |
| | 10 | 91 | | } |
| | | 92 | | |
| | 14 | 93 | | var arguments = new Dictionary<string, object?>(StringComparer.Ordinal) |
| | 14 | 94 | | { |
| | 14 | 95 | | ["x-dead-letter-exchange"] = deadLetterExchange |
| | 14 | 96 | | }; |
| | | 97 | | |
| | 14 | 98 | | if (!string.IsNullOrWhiteSpace(options.DeadLetterRoutingKey)) |
| | 8 | 99 | | arguments["x-dead-letter-routing-key"] = options.DeadLetterRoutingKey; |
| | | 100 | | |
| | 14 | 101 | | return arguments; |
| | 633 | 102 | | } |
| | | 103 | | |
| | | 104 | | /// <summary>Creates the requested resource.</summary> |
| | | 105 | | public static BasicProperties CreatePersistentJsonProperties(string? correlationId, string correlationHeader) |
| | | 106 | | { |
| | 463 | 107 | | var properties = new BasicProperties |
| | 463 | 108 | | { |
| | 463 | 109 | | ContentType = "application/json", |
| | 463 | 110 | | DeliveryMode = DeliveryModes.Persistent, |
| | 463 | 111 | | Persistent = true, |
| | 463 | 112 | | MessageId = Guid.NewGuid().ToString("N"), |
| | 463 | 113 | | Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) |
| | 463 | 114 | | }; |
| | | 115 | | |
| | 463 | 116 | | if (!string.IsNullOrWhiteSpace(correlationId)) |
| | | 117 | | { |
| | 135 | 118 | | properties.CorrelationId = correlationId; |
| | 135 | 119 | | if (!string.IsNullOrWhiteSpace(correlationHeader)) |
| | | 120 | | { |
| | 133 | 121 | | properties.Headers = new Dictionary<string, object?>(StringComparer.Ordinal) |
| | 133 | 122 | | { |
| | 133 | 123 | | [correlationHeader] = correlationId |
| | 133 | 124 | | }; |
| | | 125 | | } |
| | | 126 | | } |
| | | 127 | | |
| | 463 | 128 | | return properties; |
| | | 129 | | } |
| | | 130 | | } |