| | | 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 | | { |
| | 3 | 15 | | if (!options.DeclareTopology) |
| | 3 | 16 | | return Task.CompletedTask; |
| | | 17 | | |
| | 3 | 18 | | var exchange = RabbitMqOptionsValidator.Required(options.WorkerExchange, nameof(options.WorkerExchange)); |
| | 3 | 19 | | var queue = RabbitMqOptionsValidator.Required(options.WorkerQueue, nameof(options.WorkerQueue)); |
| | 3 | 20 | | var routingKey = RabbitMqOptionsValidator.Required(options.WorkerRoutingKey, nameof(options.WorkerRoutingKey)); |
| | | 21 | | |
| | 3 | 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 | | { |
| | 3 | 31 | | if (!options.DeclareTopology) |
| | 3 | 32 | | return Task.CompletedTask; |
| | | 33 | | |
| | 3 | 34 | | var exchange = RabbitMqOptionsValidator.Required(options.ResponseExchange, nameof(options.ResponseExchange)); |
| | 3 | 35 | | var queue = RabbitMqOptionsValidator.Required(options.ResponseQueue, nameof(options.ResponseQueue)); |
| | 3 | 36 | | var routingKey = RabbitMqOptionsValidator.Required(options.ResponseRoutingKey, nameof(options.ResponseRoutingKey |
| | | 37 | | |
| | 3 | 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 | | { |
| | 3 | 49 | | await channel.ExchangeDeclareAsync(exchange, DirectExchange, durable: true, autoDelete: false, cancellationToken |
| | | 50 | | |
| | 3 | 51 | | var queueArguments = await EnsureDeadLetterTopologyAsync(channel, options, routingKey, cancellationToken).Config |
| | | 52 | | |
| | 3 | 53 | | await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false, queueArguments, cance |
| | 3 | 54 | | await channel.QueueBindAsync(queue, exchange, routingKey, cancellationToken).ConfigureAwait(false); |
| | 3 | 55 | | } |
| | | 56 | | |
| | | 57 | | /// <summary> |
| | | 58 | | /// Declares the dead-letter exchange/queue/binding when configured and returns the arguments to apply to the |
| | | 59 | | /// source queue (<c>x-dead-letter-exchange</c> and optionally <c>x-dead-letter-routing-key</c>), or null when |
| | | 60 | | /// no dead-letter exchange is configured. |
| | | 61 | | /// </summary> |
| | | 62 | | private static async Task<IDictionary<string, object?>?> EnsureDeadLetterTopologyAsync( |
| | | 63 | | IRabbitMqChannel channel, |
| | | 64 | | RabbitMqAsyncResponseOptions options, |
| | | 65 | | string sourceRoutingKey, |
| | | 66 | | CancellationToken cancellationToken) |
| | | 67 | | { |
| | 3 | 68 | | if (string.IsNullOrWhiteSpace(options.DeadLetterExchange)) |
| | 3 | 69 | | return null; |
| | | 70 | | |
| | 2 | 71 | | var deadLetterExchange = options.DeadLetterExchange; |
| | 2 | 72 | | await channel.ExchangeDeclareAsync(deadLetterExchange, DirectExchange, durable: true, autoDelete: false, cancell |
| | | 73 | | |
| | 2 | 74 | | if (!string.IsNullOrWhiteSpace(options.DeadLetterQueue)) |
| | | 75 | | { |
| | 2 | 76 | | var bindingRoutingKey = string.IsNullOrWhiteSpace(options.DeadLetterRoutingKey) |
| | 2 | 77 | | ? sourceRoutingKey |
| | 2 | 78 | | : options.DeadLetterRoutingKey; |
| | | 79 | | |
| | 2 | 80 | | await channel.QueueDeclareAsync(options.DeadLetterQueue, durable: true, exclusive: false, autoDelete: false, |
| | 2 | 81 | | await channel.QueueBindAsync(options.DeadLetterQueue, deadLetterExchange, bindingRoutingKey, cancellationTok |
| | 2 | 82 | | } |
| | | 83 | | |
| | 2 | 84 | | var arguments = new Dictionary<string, object?>(StringComparer.Ordinal) |
| | 2 | 85 | | { |
| | 2 | 86 | | ["x-dead-letter-exchange"] = deadLetterExchange |
| | 2 | 87 | | }; |
| | | 88 | | |
| | 2 | 89 | | if (!string.IsNullOrWhiteSpace(options.DeadLetterRoutingKey)) |
| | 2 | 90 | | arguments["x-dead-letter-routing-key"] = options.DeadLetterRoutingKey; |
| | | 91 | | |
| | 2 | 92 | | return arguments; |
| | 3 | 93 | | } |
| | | 94 | | |
| | | 95 | | /// <summary>Creates the requested resource.</summary> |
| | | 96 | | public static BasicProperties CreatePersistentJsonProperties(string? correlationId, string correlationHeader) |
| | | 97 | | { |
| | 3 | 98 | | var properties = new BasicProperties |
| | 3 | 99 | | { |
| | 3 | 100 | | ContentType = "application/json", |
| | 3 | 101 | | DeliveryMode = DeliveryModes.Persistent, |
| | 3 | 102 | | Persistent = true, |
| | 3 | 103 | | MessageId = Guid.NewGuid().ToString("N"), |
| | 3 | 104 | | Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) |
| | 3 | 105 | | }; |
| | | 106 | | |
| | 3 | 107 | | if (!string.IsNullOrWhiteSpace(correlationId)) |
| | | 108 | | { |
| | 3 | 109 | | properties.CorrelationId = correlationId; |
| | 3 | 110 | | if (!string.IsNullOrWhiteSpace(correlationHeader)) |
| | | 111 | | { |
| | 3 | 112 | | properties.Headers = new Dictionary<string, object?>(StringComparer.Ordinal) |
| | 3 | 113 | | { |
| | 3 | 114 | | [correlationHeader] = correlationId |
| | 3 | 115 | | }; |
| | | 116 | | } |
| | | 117 | | } |
| | | 118 | | |
| | 3 | 119 | | return properties; |
| | | 120 | | } |
| | | 121 | | } |