Skip to content

Commit 164dd20

Browse files
pwelter34pwelter34
committed
Add ServiceBus messaging project and tests
1 parent 5594b86 commit 164dd20

13 files changed

Lines changed: 1402 additions & 0 deletions

‎Arbiter.slnx‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
<Project Path="test/Arbiter.Dispatcher.Client.Tests/Arbiter.Dispatcher.Client.Tests.csproj" Id="a27a225b-7be5-4fa8-9023-690f07312a86" />
4949
<Project Path="test/Arbiter.Mapping.Tests/Arbiter.Mapping.Tests.csproj" Id="c74c00ab-3fc8-44dc-87d3-a6e618f53c8f" />
5050
<Project Path="test/Arbiter.Mediation.Tests/Arbiter.Mediation.Tests.csproj" />
51+
<Project Path="test/Arbiter.Messaging.ServiceBus.Tests/Arbiter.Messaging.ServiceBus.Tests.csproj" Id="65e5b4e8-9c88-41ee-9487-80c378a18133" />
5152
<Project Path="test/Arbiter.OpenTelemetry.Monitor.Tests/Arbiter.OpenTelemetry.Monitor.Tests.csproj" />
5253
<Project Path="test/Arbiter.Services.Tests/Arbiter.Services.Tests.csproj" Id="f3d73b4c-09de-4369-b45a-eba1aac348ed" />
5354
</Folder>
@@ -61,5 +62,6 @@
6162
<Project Path="src/Arbiter.Mapping.Generators/Arbiter.Mapping.Generators.csproj" Id="f6265a63-6ba9-46f3-a6ea-8b6ba080487e" />
6263
<Project Path="src/Arbiter.Mapping/Arbiter.Mapping.csproj" Id="394985d4-860d-41b0-bc51-89995645d1cd" />
6364
<Project Path="src/Arbiter.Mediation/Arbiter.Mediation.csproj" />
65+
<Project Path="src/Arbiter.Messaging.ServiceBus/Arbiter.Messaging.ServiceBus.csproj" Id="5e058d58-f563-461b-b0e5-efd7b67b699f" />
6466
<Project Path="src/Arbiter.Services/Arbiter.Services.csproj" Id="d25d1a52-81b3-4a16-9a77-415950b0cac3" />
6567
</Solution>

‎Directory.Packages.props‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
<PackageVersion Include="Azure.Communication.Email" Version="1.1.0" />
1313
<PackageVersion Include="Azure.Communication.Sms" Version="1.0.2" />
1414
<PackageVersion Include="Azure.Identity" Version="1.21.0" />
15+
<PackageVersion Include="Azure.Messaging.ServiceBus" Version="7.20.1" />
1516
<PackageVersion Include="Azure.Monitor.Query" Version="1.7.1" />
1617
<PackageVersion Include="BenchmarkDotNet" Version="0.15.8" />
1718
<PackageVersion Include="Blazilla" Version="2.3.2" />
@@ -44,6 +45,7 @@
4445
<PackageVersion Include="Microsoft.Extensions.Diagnostics.HealthChecks" Version="10.0.8" />
4546
<PackageVersion Include="Microsoft.Extensions.Diagnostics.HealthChecks.EntityFrameworkCore" Version="10.0.8" />
4647
<PackageVersion Include="Microsoft.Extensions.Hosting" Version="10.0.8" />
48+
<PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.8" />
4749
<PackageVersion Include="Microsoft.Extensions.Http" Version="10.0.8" />
4850
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.8" />
4951
<PackageVersion Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.8" />

‎src/Arbiter.Messaging.ServiceBus/Arbiter.Messaging.ServiceBus.csproj‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<TargetFrameworks>net8.0;net9.0;net10.0</TargetFrameworks>
5+
<Description>Azure Service Bus messaging extensions with queue, topic, subscription initialization and sender helpers</Description>
6+
</PropertyGroup>
7+
8+
<ItemGroup>
9+
<PackageReference Include="Azure.Messaging.ServiceBus" />
10+
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" />
11+
</ItemGroup>
12+
13+
<ItemGroup>
14+
<ProjectReference Include="..\Arbiter.CommandQuery\Arbiter.CommandQuery.csproj" />
15+
</ItemGroup>
16+
17+
</Project>

‎src/Arbiter.Messaging.ServiceBus/ServiceBusExtensions.cs‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
using Azure.Messaging.ServiceBus;
2+
3+
using Microsoft.Extensions.Configuration;
4+
using Microsoft.Extensions.DependencyInjection;
5+
using Microsoft.Extensions.DependencyInjection.Extensions;
6+
7+
namespace Arbiter.Messaging.ServiceBus;
8+
9+
/// <summary>
10+
/// Provides extension methods for registering Azure Service Bus services.
11+
/// </summary>
12+
public static class ServiceBusExtensions
13+
{
14+
/// <summary>
15+
/// Registers a named Azure Service Bus client, configured senders, and resource initializer.
16+
/// </summary>
17+
/// <param name="services">The service collection to add registrations to.</param>
18+
/// <param name="serviceName">The name used to register and resolve the <see cref="ServiceBusClient" />.</param>
19+
/// <param name="nameOrConnectionString">A Service Bus connection string, connection string name, or configuration key.</param>
20+
/// <param name="configure">A delegate used to configure queues, topics, subscriptions, and options.</param>
21+
/// <returns>The same <see cref="IServiceCollection" /> instance for chaining.</returns>
22+
public static IServiceCollection AddServiceBus(
23+
this IServiceCollection services,
24+
object? serviceName,
25+
string nameOrConnectionString,
26+
Action<ServiceBusOptions> configure)
27+
{
28+
ArgumentNullException.ThrowIfNull(services);
29+
ArgumentException.ThrowIfNullOrWhiteSpace(nameOrConnectionString);
30+
ArgumentNullException.ThrowIfNull(configure);
31+
32+
var options = new ServiceBusOptions
33+
{
34+
ServiceKey = serviceName,
35+
NameOrConnectionString = nameOrConnectionString,
36+
};
37+
38+
configure.Invoke(options);
39+
40+
// register for enumeration and as keyed for resolution
41+
services.AddSingleton(options);
42+
services.TryAddKeyedSingleton(serviceName, options);
43+
44+
services.TryAddKeyedSingleton(
45+
serviceKey: serviceName,
46+
implementationFactory: (sp, _) =>
47+
{
48+
var connectionString = sp.ResolveConnectionString(options.NameOrConnectionString);
49+
return new ServiceBusClient(connectionString);
50+
}
51+
);
52+
53+
// register senders for queues and topics
54+
foreach (var queue in options.Queues.Concat(options.Topics))
55+
{
56+
services.AddKeyedSingleton(queue, (sp, _) =>
57+
{
58+
// if NameSuffix is provided, append to queue/topic name for sender resolution
59+
var queueName = options.FormatName(queue);
60+
var serviceBusClient = sp.GetRequiredKeyedService<ServiceBusClient>(options.ServiceKey);
61+
return serviceBusClient.CreateSender(queueName);
62+
});
63+
}
64+
65+
services.AddHostedService<ServiceBusInitializer>();
66+
67+
return services;
68+
}
69+
70+
/// <summary>
71+
/// Resolves a connection string from either a direct connection string or a configuration key name.
72+
/// </summary>
73+
/// <param name="services">The service provider used to resolve application configuration.</param>
74+
/// <param name="nameOrConnectionString">A direct connection string, connection string name, or configuration key.</param>
75+
/// <returns>The resolved connection string, or the original value when no configuration value is found.</returns>
76+
public static string ResolveConnectionString(
77+
this IServiceProvider services,
78+
string nameOrConnectionString)
79+
{
80+
ArgumentNullException.ThrowIfNull(services);
81+
ArgumentException.ThrowIfNullOrWhiteSpace(nameOrConnectionString);
82+
83+
var isConnectionString = nameOrConnectionString.IndexOfAny([';', '=', ':', '/']) > 0;
84+
if (isConnectionString)
85+
return nameOrConnectionString;
86+
87+
var configuration = services.GetRequiredService<IConfiguration>();
88+
89+
// first try connection strings section
90+
var connectionString = configuration.GetConnectionString(nameOrConnectionString);
91+
if (!string.IsNullOrEmpty(connectionString))
92+
return connectionString!;
93+
94+
// next try root collection
95+
connectionString = configuration[nameOrConnectionString];
96+
if (!string.IsNullOrEmpty(connectionString))
97+
return connectionString!;
98+
99+
return nameOrConnectionString;
100+
}
101+
102+
}

‎src/Arbiter.Messaging.ServiceBus/ServiceBusInitializer.cs‎

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
using Azure;
2+
using Azure.Messaging.ServiceBus.Administration;
3+
4+
using Microsoft.Extensions.DependencyInjection;
5+
using Microsoft.Extensions.Hosting;
6+
using Microsoft.Extensions.Logging;
7+
8+
namespace Arbiter.Messaging.ServiceBus;
9+
10+
/// <summary>
11+
/// Initializes configured Azure Service Bus queues, topics, and subscriptions when the host starts.
12+
/// </summary>
13+
public sealed class ServiceBusInitializer : IHostedService
14+
{
15+
private readonly ILogger<ServiceBusInitializer> _logger;
16+
private readonly IServiceProvider _serviceProvider;
17+
18+
/// <summary>
19+
/// Initializes a new instance of the <see cref="ServiceBusInitializer" /> class.
20+
/// </summary>
21+
/// <param name="logger">The logger used to write initialization messages.</param>
22+
/// <param name="serviceProvider">The service provider used to resolve configured options and dependencies.</param>
23+
public ServiceBusInitializer(ILogger<ServiceBusInitializer> logger, IServiceProvider serviceProvider)
24+
{
25+
ArgumentNullException.ThrowIfNull(logger);
26+
ArgumentNullException.ThrowIfNull(serviceProvider);
27+
28+
_logger = logger;
29+
_serviceProvider = serviceProvider;
30+
}
31+
32+
33+
/// <summary>
34+
/// Starts Service Bus resource initialization.
35+
/// </summary>
36+
/// <param name="cancellationToken">A token used to cancel initialization.</param>
37+
/// <returns>A task that represents the asynchronous start operation.</returns>
38+
public async Task StartAsync(CancellationToken cancellationToken)
39+
{
40+
var options = _serviceProvider.GetServices<ServiceBusOptions>();
41+
if (options?.Any() != true)
42+
{
43+
_logger.LogDebug("No Azure Service Bus resources configured for initialization.");
44+
return;
45+
}
46+
47+
foreach (var option in options)
48+
{
49+
try
50+
{
51+
await Initialize(option, cancellationToken).ConfigureAwait(false);
52+
}
53+
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
54+
{
55+
throw;
56+
}
57+
catch (Exception ex)
58+
{
59+
_logger.LogError(ex, "Error initializing Azure Service Bus resources for '{NameOrConnectionString}': {ErrorMessage}", option.NameOrConnectionString, ex.Message);
60+
}
61+
}
62+
}
63+
64+
/// <summary>
65+
/// Stops the hosted service.
66+
/// </summary>
67+
/// <param name="cancellationToken">A token used to cancel the stop operation.</param>
68+
/// <returns>A completed task.</returns>
69+
public Task StopAsync(CancellationToken cancellationToken)
70+
{
71+
return Task.CompletedTask;
72+
}
73+
74+
75+
private async Task Initialize(
76+
ServiceBusOptions option,
77+
CancellationToken cancellationToken)
78+
{
79+
var connectionString = _serviceProvider.ResolveConnectionString(option.NameOrConnectionString);
80+
var adminClient = new ServiceBusAdministrationClient(connectionString);
81+
82+
foreach (var queue in option.Queues)
83+
await InitializeQueue(adminClient, option, queue, cancellationToken).ConfigureAwait(false);
84+
85+
foreach (var topic in option.Topics)
86+
await InitializeTopic(adminClient, option, topic, cancellationToken).ConfigureAwait(false);
87+
}
88+
89+
private async Task InitializeQueue(
90+
ServiceBusAdministrationClient adminClient,
91+
ServiceBusOptions option,
92+
string queueName,
93+
CancellationToken cancellationToken)
94+
{
95+
var queue = option.FormatName(queueName);
96+
97+
var queueExists = await adminClient.QueueExistsAsync(queue, cancellationToken).ConfigureAwait(false);
98+
if (queueExists.Value)
99+
return;
100+
101+
var options = new CreateQueueOptions(queue)
102+
{
103+
DefaultMessageTimeToLive = option.DefaultMessageTimeToLive,
104+
MaxDeliveryCount = option.MaxDeliveryCount,
105+
LockDuration = option.LockDuration,
106+
DeadLetteringOnMessageExpiration = option.DeadLetteringOnMessageExpiration,
107+
EnableBatchedOperations = option.EnableBatchedOperations,
108+
};
109+
110+
_logger.LogInformation("Creating Queue '{QueueName}'", queue);
111+
112+
try
113+
{
114+
await adminClient.CreateQueueAsync(options, cancellationToken: cancellationToken).ConfigureAwait(false);
115+
}
116+
catch (RequestFailedException ex) when (ex.Status == 409)
117+
{
118+
_logger.LogDebug(ex, "Queue '{QueueName}' already exists", queue);
119+
}
120+
}
121+
122+
private async Task InitializeTopic(
123+
ServiceBusAdministrationClient adminClient,
124+
ServiceBusOptions option,
125+
string topicName,
126+
CancellationToken cancellationToken)
127+
{
128+
var topic = option.FormatName(topicName);
129+
130+
var topicExists = await adminClient.TopicExistsAsync(topic, cancellationToken).ConfigureAwait(false);
131+
if (!topicExists.Value)
132+
{
133+
var topicOptions = new CreateTopicOptions(topic)
134+
{
135+
DefaultMessageTimeToLive = option.DefaultMessageTimeToLive,
136+
EnableBatchedOperations = option.EnableBatchedOperations,
137+
};
138+
139+
_logger.LogInformation("Creating Topic '{TopicName}'", topic);
140+
141+
try
142+
{
143+
await adminClient.CreateTopicAsync(topicOptions, cancellationToken).ConfigureAwait(false);
144+
}
145+
catch (RequestFailedException ex) when (ex.Status == 409)
146+
{
147+
_logger.LogDebug(ex, "Topic '{TopicName}' already exists", topic);
148+
}
149+
}
150+
151+
if (!option.Subscriptions.TryGetValue(topicName, out var subscriptions))
152+
return;
153+
154+
foreach (var subscriptionName in subscriptions)
155+
{
156+
var subscriptionExists = await adminClient.SubscriptionExistsAsync(topic, subscriptionName, cancellationToken).ConfigureAwait(false);
157+
if (subscriptionExists.Value)
158+
continue;
159+
160+
var subscriptionOptions = new CreateSubscriptionOptions(topic, subscriptionName)
161+
{
162+
DefaultMessageTimeToLive = option.DefaultMessageTimeToLive,
163+
MaxDeliveryCount = option.MaxDeliveryCount,
164+
LockDuration = option.LockDuration,
165+
DeadLetteringOnMessageExpiration = option.DeadLetteringOnMessageExpiration,
166+
EnableBatchedOperations = option.EnableBatchedOperations,
167+
};
168+
169+
_logger.LogInformation("Creating Subscription '{SubscriptionName}' for Topic '{TopicName}'", subscriptionName, topic);
170+
171+
try
172+
{
173+
await adminClient.CreateSubscriptionAsync(subscriptionOptions, cancellationToken).ConfigureAwait(false);
174+
}
175+
catch (RequestFailedException ex) when (ex.Status == 409)
176+
{
177+
_logger.LogDebug(ex, "Subscription '{SubscriptionName}' for Topic '{TopicName}' already exists", subscriptionName, topic);
178+
}
179+
}
180+
}
181+
}

0 commit comments

Comments
 (0)