| | | 1 | | using Azure.Messaging.ServiceBus; |
| | | 2 | | using Microsoft.Extensions.DependencyInjection; |
| | | 3 | | |
| | | 4 | | namespace AsyncResponse.Transports.AzureServiceBus; |
| | | 5 | | |
| | | 6 | | internal interface IAzureServiceBusClient : IAsyncDisposable |
| | | 7 | | { |
| | | 8 | | IAzureServiceBusSender CreateSender(string queue); |
| | | 9 | | IAzureServiceBusReceiver CreateReceiver(string queue, AzureServiceBusSubscriberOptions subscriberOptions); |
| | | 10 | | } |
| | | 11 | | |
| | | 12 | | internal sealed class AzureServiceBusClientAdapter( |
| | | 13 | | ServiceBusClient inner, |
| | | 14 | | bool ownsClient) : IAzureServiceBusClient |
| | | 15 | | { |
| | | 16 | | /// <summary>Creates a sender for the requested queue.</summary> |
| | | 17 | | public IAzureServiceBusSender CreateSender(string queue) |
| | | 18 | | => new AzureServiceBusSenderAdapter(inner.CreateSender(queue)); |
| | | 19 | | |
| | | 20 | | /// <summary>Creates a peek-lock receiver for the requested queue.</summary> |
| | | 21 | | public IAzureServiceBusReceiver CreateReceiver( |
| | | 22 | | string queue, |
| | | 23 | | AzureServiceBusSubscriberOptions subscriberOptions) |
| | | 24 | | => new AzureServiceBusReceiverAdapter(inner.CreateReceiver( |
| | | 25 | | queue, |
| | | 26 | | new ServiceBusReceiverOptions |
| | | 27 | | { |
| | | 28 | | ReceiveMode = ServiceBusReceiveMode.PeekLock, |
| | | 29 | | PrefetchCount = subscriberOptions.PrefetchCount |
| | | 30 | | })); |
| | | 31 | | |
| | | 32 | | /// <summary>Releases resources held by this instance.</summary> |
| | | 33 | | public async ValueTask DisposeAsync() |
| | | 34 | | { |
| | | 35 | | if (ownsClient) |
| | | 36 | | await inner.DisposeAsync().ConfigureAwait(false); |
| | | 37 | | } |
| | | 38 | | } |
| | | 39 | | |
| | | 40 | | internal static class AzureServiceBusClientResolver |
| | | 41 | | { |
| | | 42 | | public static IAzureServiceBusClient Create(IServiceProvider provider) |
| | | 43 | | { |
| | | 44 | | if (provider.GetService<ServiceBusClient>() is { } registeredClient) |
| | | 45 | | return new AzureServiceBusClientAdapter(registeredClient, ownsClient: false); |
| | | 46 | | |
| | | 47 | | var options = provider.GetRequiredService<Microsoft.Extensions.Options.IOptions<AzureServiceBusAsyncResponseOpti |
| | | 48 | | var connectionString = AzureServiceBusOptionsValidator.Required(options.ConnectionString, nameof(options.Connect |
| | | 49 | | return new AzureServiceBusClientAdapter(new ServiceBusClient(connectionString), ownsClient: true); |
| | | 50 | | } |
| | | 51 | | } |
| | | 52 | | |
| | | 53 | | internal interface IAzureServiceBusSender : IAsyncDisposable |
| | | 54 | | { |
| | | 55 | | Task SendMessageAsync(AzureServiceBusOutboundMessage message, CancellationToken cancellationToken = default); |
| | | 56 | | Task CloseAsync(CancellationToken cancellationToken = default); |
| | | 57 | | } |
| | | 58 | | |
| | | 59 | | internal sealed class AzureServiceBusSenderAdapter(ServiceBusSender inner) : IAzureServiceBusSender |
| | | 60 | | { |
| | | 61 | | /// <summary>Sends the supplied outbound message.</summary> |
| | | 62 | | public Task SendMessageAsync(AzureServiceBusOutboundMessage message, CancellationToken cancellationToken = default) |
| | | 63 | | { |
| | | 64 | | var serviceBusMessage = new ServiceBusMessage(BinaryData.FromString(message.Body)) |
| | | 65 | | { |
| | | 66 | | ContentType = "application/json", |
| | | 67 | | MessageId = message.MessageId, |
| | | 68 | | CorrelationId = message.CorrelationId |
| | | 69 | | }; |
| | | 70 | | |
| | | 71 | | foreach (var property in message.ApplicationProperties) |
| | | 72 | | serviceBusMessage.ApplicationProperties[property.Key] = property.Value; |
| | | 73 | | |
| | | 74 | | return inner.SendMessageAsync(serviceBusMessage, cancellationToken); |
| | | 75 | | } |
| | | 76 | | |
| | | 77 | | /// <summary>Closes the sender link.</summary> |
| | | 78 | | public Task CloseAsync(CancellationToken cancellationToken = default) |
| | | 79 | | => inner.CloseAsync(cancellationToken); |
| | | 80 | | |
| | | 81 | | /// <summary>Releases resources held by this instance.</summary> |
| | | 82 | | public ValueTask DisposeAsync() => inner.DisposeAsync(); |
| | | 83 | | } |
| | | 84 | | |
| | | 85 | | internal interface IAzureServiceBusReceiver : IAsyncDisposable |
| | | 86 | | { |
| | | 87 | | Task<IReadOnlyList<AzureServiceBusTransportDelivery>> ReceiveMessagesAsync( |
| | | 88 | | int maxMessages, |
| | | 89 | | TimeSpan maxWaitTime, |
| | | 90 | | CancellationToken cancellationToken = default); |
| | | 91 | | |
| | | 92 | | Task CloseAsync(CancellationToken cancellationToken = default); |
| | | 93 | | } |
| | | 94 | | |
| | 3 | 95 | | internal sealed class AzureServiceBusReceiverAdapter( |
| | 3 | 96 | | ServiceBusReceiver inner, |
| | 3 | 97 | | string? queueOverride = null) : IAzureServiceBusReceiver |
| | | 98 | | { |
| | | 99 | | /// <summary>Receives and wraps messages from Service Bus.</summary> |
| | | 100 | | public async Task<IReadOnlyList<AzureServiceBusTransportDelivery>> ReceiveMessagesAsync( |
| | | 101 | | int maxMessages, |
| | | 102 | | TimeSpan maxWaitTime, |
| | | 103 | | CancellationToken cancellationToken = default) |
| | | 104 | | { |
| | 3 | 105 | | var messages = await inner.ReceiveMessagesAsync(maxMessages, maxWaitTime, cancellationToken).ConfigureAwait(fals |
| | 3 | 106 | | if (messages.Count == 0) |
| | 3 | 107 | | return []; |
| | | 108 | | |
| | 3 | 109 | | var queue = queueOverride ?? inner.EntityPath; |
| | 3 | 110 | | var deliveries = new AzureServiceBusTransportDelivery[messages.Count]; |
| | 3 | 111 | | for (var i = 0; i < messages.Count; i++) |
| | 3 | 112 | | deliveries[i] = CreateDelivery(queue, messages[i]); |
| | | 113 | | |
| | 3 | 114 | | return deliveries; |
| | 3 | 115 | | } |
| | | 116 | | |
| | | 117 | | private AzureServiceBusTransportDelivery CreateDelivery( |
| | | 118 | | string queue, |
| | | 119 | | ServiceBusReceivedMessage message) |
| | 3 | 120 | | => new( |
| | 3 | 121 | | queue, |
| | 3 | 122 | | message.Body.ToString(), |
| | 3 | 123 | | message.MessageId, |
| | 3 | 124 | | message.CorrelationId, |
| | 3 | 125 | | message.SequenceNumber, |
| | 3 | 126 | | message.DeliveryCount, |
| | 3 | 127 | | new Dictionary<string, object?>(message.ApplicationProperties, StringComparer.OrdinalIgnoreCase), |
| | 3 | 128 | | () => new ValueTask(inner.CompleteMessageAsync(message, CancellationToken.None)), |
| | 2 | 129 | | () => new ValueTask(inner.AbandonMessageAsync(message, cancellationToken: CancellationToken.None)), |
| | 2 | 130 | | (reason, description) => new ValueTask(inner.DeadLetterMessageAsync( |
| | 2 | 131 | | message, |
| | 2 | 132 | | deadLetterReason: reason, |
| | 2 | 133 | | deadLetterErrorDescription: description, |
| | 2 | 134 | | cancellationToken: CancellationToken.None)), |
| | 3 | 135 | | // Settlement deliberately ignores cancellation so an in-flight message still settles |
| | 3 | 136 | | // during shutdown. Lock renewal is a background courtesy and honors the caller's token: |
| | 3 | 137 | | // on a degraded namespace each renew otherwise burns the SDK's full retry budget, and |
| | 3 | 138 | | // the renewal loop must be interruptible mid-call for the batch (and shutdown) to |
| | 3 | 139 | | // complete promptly. |
| | 3 | 140 | | cancellationToken => new ValueTask(inner.RenewMessageLockAsync(message, cancellationToken))); |
| | | 141 | | |
| | | 142 | | /// <summary>Closes the receiver link.</summary> |
| | | 143 | | public Task CloseAsync(CancellationToken cancellationToken = default) |
| | 3 | 144 | | => inner.CloseAsync(cancellationToken); |
| | | 145 | | |
| | | 146 | | /// <summary>Releases resources held by this instance.</summary> |
| | 3 | 147 | | public ValueTask DisposeAsync() => inner.DisposeAsync(); |
| | | 148 | | } |
| | | 149 | | |
| | | 150 | | internal sealed record AzureServiceBusOutboundMessage( |
| | | 151 | | string Body, |
| | | 152 | | string MessageId, |
| | | 153 | | string? CorrelationId, |
| | | 154 | | IReadOnlyDictionary<string, object?> ApplicationProperties); |
| | | 155 | | |
| | | 156 | | internal sealed record AzureServiceBusTransportDelivery( |
| | | 157 | | string Queue, |
| | | 158 | | string Body, |
| | | 159 | | string MessageId, |
| | | 160 | | string? CorrelationId, |
| | | 161 | | long SequenceNumber, |
| | | 162 | | int DeliveryCount, |
| | | 163 | | IReadOnlyDictionary<string, object?> ApplicationProperties, |
| | | 164 | | Func<ValueTask> CompleteAsync, |
| | | 165 | | Func<ValueTask> AbandonAsync, |
| | | 166 | | Func<string, string?, ValueTask> DeadLetterAsync, |
| | | 167 | | Func<CancellationToken, ValueTask> RenewLockAsync); |