Skip to content

Commit

Permalink
Extend CAP.AzureServiceBusOptions (#1514)
Browse files Browse the repository at this point in the history
  • Loading branch information
@vyefremov
vyefremov committed Apr 12, 2024
1 parent 394bb60 commit 31d77b0
Show file tree
Hide file tree
Showing 3 changed files with 93 additions and 17 deletions.
30 changes: 18 additions & 12 deletions docs/content/user-guide/en/transport/azure-service-bus.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,18 +39,24 @@ public void ConfigureServices(IServiceCollection services)

The AzureServiceBus configuration options provided directly by the CAP:

| NAME | DESCRIPTION | TYPE | DEFAULT |
|:------------------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------|:----------------------------------|
| ConnectionString | Endpoint address | string | |
| EnableSessions | Enable [Service bus sessions](https://docs.microsoft.com/en-us/azure/service-bus-messaging/message-sessions) | bool | false |
| TopicPath | Topic entity path | string | cap |
| SubscriptionAutoDeleteOnIdle | Automatically delete subscription after a certain idle interval. | TimeSpan | TimeSpan.MaxValue |
| ManagementTokenProvider | Token provider | ITokenProvider | null |
| AutoCompleteMessages | Gets a value that indicates whether the processor should automatically complete messages after the message handler has completed processing | bool | false |
| CustomHeaders | Adds custom and/or mandatory Headers for incoming messages from heterogeneous systems. | `Func<Message, List<KeyValuePair<string, string>>>?` | null |
| Namespace | Namespace of Servicebus , Needs to be set when using with TokenCredential Property | string | null |
| DefaultCorrelationHeaders | Adds additional correlation properties to all [correlation filters](https://learn.microsoft.com/en-us/azure/service-bus-messaging/topic-filters#correlation-filters). | IDictionary<string, string> | Dictionary<string, string>.Empty |
| SQLFilters | Custom SQL Filters by name and expression on Topic Subscribtion | List<KeyValuePair<string, string>> | null |
| NAME | DESCRIPTION | TYPE | DEFAULT |
|:-------------------------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------|:---------------------------------|
| ConnectionString | Endpoint address | string | |
| TopicPath | Topic entity path | string | cap |
| EnableSessions | Enable [Service bus sessions](https://docs.microsoft.com/en-us/azure/service-bus-messaging/message-sessions) | bool | false |
| MaxConcurrentSessions | The maximum number of concurrent sessions that the processor can handle. Not applicable when EnableSessions is false. | int | 8 |
| SessionIdleTimeout | The maximum time to wait for a new message before the session is closed. If not specified, 60 seconds will be used by Azure Service Bus. | TimeSpan | null |
| SubscriptionAutoDeleteOnIdle | Automatically delete subscription after a certain idle interval. | TimeSpan | TimeSpan.MaxValue |
| SubscriptionMessageLockDuration | The amount of time the message is locked by a given receiver so that no other receiver receives the same message. | TimeSpan | 60 seconds |
| SubscriptionDefaultMessageTimeToLive | The default message time to live value for a subscription. This is the duration after which the message expires. | TimeSpan | TimeSpan.MaxValue |
| SubscriptionMaxDeliveryCount | The maximum number of times a message is delivered to the subscription before it is dead-lettered. | int | 10 |
| MaxAutoLockRenewalDuration | The maximum duration within which the lock will be renewed automatically. This value should be greater than the longest message lock duration. | TimeSpan | 5 minutes |
| ManagementTokenProvider | Token provider | ITokenProvider | null |
| AutoCompleteMessages | Gets a value that indicates whether the processor should automatically complete messages after the message handler has completed processing | bool | true |
| CustomHeaders | Adds custom and/or mandatory Headers for incoming messages from heterogeneous systems. | `Func<Message, List<KeyValuePair<string, string>>>?` | null |
| Namespace | Namespace of Servicebus , Needs to be set when using with TokenCredential Property | string | null |
| DefaultCorrelationHeaders | Adds additional correlation properties to all [correlation filters](https://learn.microsoft.com/en-us/azure/service-bus-messaging/topic-filters#correlation-filters). | IDictionary<string, string> | Dictionary<string, string>.Empty |
| SQLFilters | Custom SQL Filters by name and expression on Topic Subscribtion | List<KeyValuePair<string, string>> | null |

#### Sessions

Expand Down
19 changes: 16 additions & 3 deletions src/DotNetCore.CAP.AzureServiceBus/AzureServiceBusConsumerClient.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -214,26 +214,39 @@ public async Task ConnectAsync()
new CreateSubscriptionOptions(topicPath, _subscriptionName)
{
RequiresSession = _asbOptions.EnableSessions,
AutoDeleteOnIdle = _asbOptions.SubscriptionAutoDeleteOnIdle
AutoDeleteOnIdle = _asbOptions.SubscriptionAutoDeleteOnIdle,
LockDuration = _asbOptions.SubscriptionMessageLockDuration,
DefaultMessageTimeToLive = _asbOptions.SubscriptionDefaultMessageTimeToLive,
MaxDeliveryCount = _asbOptions.SubscriptionMaxDeliveryCount,
};

await _administrationClient.CreateSubscriptionAsync(subscriptionDescription);

_logger.LogInformation(
$"Azure Service Bus topic {topicPath} created subscription: {_subscriptionName}");
}
}

_serviceBusProcessor = !_asbOptions.EnableSessions
? new ServiceBusProcessorFacade(
_serviceBusClient.CreateProcessor(_asbOptions.TopicPath, _subscriptionName))
serviceBusProcessor: _serviceBusClient.CreateProcessor(_asbOptions.TopicPath,
_subscriptionName,
new ServiceBusProcessorOptions
{
AutoCompleteMessages = _asbOptions.AutoCompleteMessages,
MaxConcurrentCalls = _asbOptions.MaxConcurrentCalls,
MaxAutoLockRenewalDuration = _asbOptions.MaxAutoLockRenewalDuration,
}))
: new ServiceBusProcessorFacade(
serviceBusSessionProcessor: _serviceBusClient.CreateSessionProcessor(_asbOptions.TopicPath,
_subscriptionName,
new ServiceBusSessionProcessorOptions
{
AutoCompleteMessages = _asbOptions.AutoCompleteMessages,
MaxConcurrentCallsPerSession = _asbOptions.MaxConcurrentCalls,
MaxAutoLockRenewalDuration = TimeSpan.FromSeconds(30),
MaxAutoLockRenewalDuration = _asbOptions.MaxAutoLockRenewalDuration,
MaxConcurrentSessions = _asbOptions.MaxConcurrentSessions,
SessionIdleTimeout = _asbOptions.SessionIdleTimeout,
}));
}
}
Expand Down
61 changes: 59 additions & 2 deletions src/DotNetCore.CAP.AzureServiceBus/CAP.AzureServiceBusOptions.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

using System;
using System.Collections.Generic;
using System.Threading;
using Azure.Core;
using Azure.Messaging.ServiceBus;
using DotNetCore.CAP.AzureServiceBus;
Expand Down Expand Up @@ -48,12 +49,40 @@ public class AzureServiceBusOptions
/// <remarks>The minimum duration is 5 minutes. Default value is <see cref="TimeSpan.MaxValue" />.</remarks>
public TimeSpan SubscriptionAutoDeleteOnIdle { get; set; } = TimeSpan.MaxValue;

/// <summary>
/// Duration of a peek lock receive. i.e., the amount of time that the message is locked by a given receiver so that
/// no other receiver receives the same message.
/// </summary>
/// <remarks>Max value is 5 minutes. Default value is 60 seconds.</remarks>
public TimeSpan SubscriptionMessageLockDuration { get; set; } = TimeSpan.FromSeconds(60);

/// <summary>
/// The default time to live value for the messages. This is the duration after which the message expires.
/// </summary>
/// <remarks>
/// This is the default value used when <see cref="ServiceBusMessage.TimeToLive"/> is not set on a
/// message itself. Messages older than their TimeToLive value will expire and no longer be retained in the message store.
/// Subscribers will be unable to receive expired messages.
/// Default value is <see cref="TimeSpan.MaxValue"/>.
/// </remarks>
public TimeSpan SubscriptionDefaultMessageTimeToLive { get; set; } = TimeSpan.MaxValue;

/// <summary>
/// The maximum delivery count of a message before it is dead-lettered.
/// </summary>
/// <remarks>
/// The delivery count is increased when a message is received in <see cref="ServiceBusReceiveMode.PeekLock"/> mode
/// and didn't complete the message before the message lock expired.
/// Default value is 10. Minimum value is 1.
/// </remarks>
public int SubscriptionMaxDeliveryCount { get; set; } = 10;

/// <summary>
/// Gets a value that indicates whether the processor should automatically complete messages after the message handler has
/// completed processing.
/// If the message handler triggers an exception, the message will not be automatically completed.
/// </summary>
public bool AutoCompleteMessages { get; set; }
public bool AutoCompleteMessages { get; set; } = true;

/// <summary>
/// Adds additional correlation properties to all correlation filters.
Expand All @@ -64,7 +93,35 @@ public class AzureServiceBusOptions
/// <summary>
/// Gets the maximum number of concurrent calls to the ProcessMessageAsync message handler the processor should initiate.
/// </summary>
public int MaxConcurrentCalls { get; set; }
/// <remarks>Default values is 1.</remarks>
public int MaxConcurrentCalls { get; set; } = 1;

/// <summary>
/// The maximum amount of time to wait for a message to be received for the
/// currently active session. After this time has elapsed, the processor will close the session
/// and attempt to process another session.
/// </summary>
/// <remarks>Not applicable when <see cref="EnableSessions"/> is false.</remarks>
public TimeSpan? SessionIdleTimeout { get; set; }

/// <summary>
/// The maximum number of sessions that can be processed concurrently by the processor.
/// </summary>
/// <remarks>
/// Not applicable when <see cref="EnableSessions"/> is false.
/// The default value is 8.
/// </remarks>
public int MaxConcurrentSessions { get; set; } = 8;

/// <summary>
/// The maximum duration within which the lock will be renewed automatically.
/// </summary>
/// <remarks>
/// This value should be greater than the longest message lock duration; for example, the LockDuration Property.
/// To specify an infinite duration, use <see cref="Timeout.InfiniteTimeSpan"/>.
/// The default value is 5 minutes.
/// </remarks>
public TimeSpan MaxAutoLockRenewalDuration { get; set; } = TimeSpan.FromMinutes(5);

/// <summary>
/// Represents the Azure Active Directory token provider for Azure Managed Service Identity integration.
Expand Down

0 comments on commit 31d77b0

Please sign in to comment.