Add Event Hub Sender

All.sln

@@ -7,7 +7,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "src", "src", "{17CDA34C-429
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "IdentityServer", "IdentityServer", "{D264CAEC-2137-46D8-B637-FCE7CBE3B390}"
 EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Messaging.AzureServiceBus", "src\IdentityServer\Messaging.AzureServiceBus\Messaging.AzureServiceBus.csproj", "{C8F188CA-2671-4364-B53C-43832627C0DA}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Messaging.Azure", "src\IdentityServer\Messaging.Azure\Messaging.Azure.csproj", "{C8F188CA-2671-4364-B53C-43832627C0DA}"
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Messaging.RabbitMQ", "src\IdentityServer\Messaging.RabbitMQ\Messaging.RabbitMQ.csproj", "{9C1A424A-64A5-4BAA-BCE5-529F2148497F}"
 EndProject

Directory.Packages.props

@@ -11,14 +11,15 @@
     <PackageVersion Include="Duende.IdentityServer.Storage" Version="7.0.4" />
     <PackageVersion Include="FluentValidation" Version="11.9.0" />
     <PackageVersion Include="FluentAssertions" Version="6.12.0" />
+    <PackageVersion Include="Masstransit.EventHub" Version="8.2.2" />
     <PackageVersion Include="Newtonsoft.Json" Version="13.0.3" />
     <PackageVersion Include="Handlebars.Net" Version="2.1.6" />
     <PackageVersion Include="HotChocolate.AspNetCore" Version="13.9.0" />
     <PackageVersion Include="HotChocolate.AspNetCore.Authorization" Version="13.9.0" />
     <PackageVersion Include="IdentityModel" Version="7.0.0" />
-    <PackageVersion Include="MassTransit" Version="8.2.1" />
-    <PackageVersion Include="MassTransit.Azure.ServiceBus.Core" Version="8.2.1" />
-    <PackageVersion Include="MassTransit.RabbitMQ" Version="8.2.1" />
+    <PackageVersion Include="MassTransit" Version="8.2.2" />
+    <PackageVersion Include="MassTransit.Azure.ServiceBus.Core" Version="8.2.2" />
+    <PackageVersion Include="MassTransit.RabbitMQ" Version="8.2.2" />
     <PackageVersion Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="8.0.4" />
     <PackageVersion Include="Microsoft.AspNetCore.Authentication.OpenIdConnect" Version="8.0.4" />
     <PackageVersion Include="Microsoft.AspNetCore.Mvc.Razor.RuntimeCompilation" Version="8.0.4" />

src/IdentityServer/Messaging.Azure/EventHubSender.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using IdOps.IdentityServer.Abstractions;
+using IdOps.Messages;
+using MassTransit;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+
+namespace IdOps.IdentityServer.Azure;
+
+public sealed class EventHubSender : BackgroundService, IEventSenderWorker
+{
+    private readonly IServiceProvider _serviceProvider;
+    private readonly ChannelReader<IdentityEventMessage> _channelReader;
+
+    public EventHubSender(
+        IServiceProvider serviceProvider,
+        ChannelReader<IdentityEventMessage> channelReader)
+    {
+        _serviceProvider = serviceProvider;
+        _channelReader = channelReader;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await Task.Yield();
+        await using AsyncServiceScope scope = _serviceProvider.CreateAsyncScope();
+        IEventHubProducerProvider provider =
+            scope.ServiceProvider.GetRequiredService<IEventHubProducerProvider>();
+
+        IEventHubProducer producer = await provider.GetProducer("identity-events");
+
+        // we reuse the buffer to avoid allocations
+        var buffer = new IdentityEventMessage[50];
+        try
+        {
+            while (await _channelReader.WaitToReadAsync(stoppingToken))
+            {
+                // we read as many messages as we can
+                for (var i = 0; i < buffer.Length; i++)
+                {
+                    if (!_channelReader.TryRead(out IdentityEventMessage? entity))
+                    {
+                        break;
+                    }
+
+                    buffer[i] = entity;
+                }
+
+                // create a batch of messages to send
+                var batch = new IdentityEventMessage[buffer.Length];
+                Array.Copy(buffer, batch, buffer.Length);
+                buffer.AsSpan().Clear();
+
+                IdOpsMeters.RecordSenderBatchSize(batch.Length);
+
+                await producer.Produce<IdentityEventMessage>(batch, stoppingToken);
+            }
+        }
+        catch
+        {
+            // ignored
+        }
+    }
+}

src/IdentityServer/Messaging.Azure/Extensions/AzureServiceBusIdOpsBuilderExtensions.cs

@@ -0,0 +1,101 @@
+using System;
+using Azure.Identity;
+using IdOps.IdentityServer.Abstractions;
+using MassTransit;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using static IdOps.IdentityServer.Wellknown.ConfigSections;
+
+namespace IdOps.IdentityServer.Azure
+{
+    public static class AzureServiceBusIdOpsBuilderExtensions
+    {
+        public static IIdOpsIdentityServerBuilder UseAzure(this BusBuilder builder)
+        {
+            AzureOptions? options = builder
+                .IdOpsBuilder.Configuration?.GetSection($"{Messaging}:Azure")
+                .Get<AzureOptions>();
+
+            if (options == null)
+            {
+                throw new ApplicationException(
+                    "Could not get AzureOptions configuration from "
+                        + $"{Messaging}:Azure."
+                        + "Please check you configuration");
+            }
+
+            return builder.UseAzure(options);
+        }
+
+        private static IIdOpsIdentityServerBuilder UseAzure(
+            this BusBuilder builder,
+            AzureOptions options)
+        {
+            if (options.EventHub is not null)
+            {
+                builder.IdOpsBuilder.Services.AddSingleton<IEventSenderWorker, EventHubSender>();
+            }
+
+            builder.IdOpsBuilder.Services.AddMassTransit(s =>
+            {
+                builder.BusSetup?.Invoke(s);
+
+                if (options.ServiceBus is { })
+                {
+                    s.RegisterServiceBus(options.ServiceBus, builder);
+                }
+
+                if (options.EventHub is { } eventHub)
+                {
+                    s.RegisterEventHub(eventHub);
+                }
+            });
+
+            return builder.IdOpsBuilder;
+        }
+
+        private static void RegisterEventHub(
+            this IBusRegistrationConfigurator configurator,
+            EventHubOptions eventHub)
+        {
+            configurator.AddRider(x =>
+                x.UsingEventHub((_, k) =>
+                {
+                    if (eventHub.Namespace is { } @namespace)
+                    {
+                        k.Host(@namespace, new DefaultAzureCredential());
+                    }
+                    else if (eventHub.ConnectionString is not null)
+                    {
+                        k.Host(eventHub.ConnectionString);
+                    }
+                    else
+                    {
+                        throw new ApplicationException(
+                            "EventHub configuration is missing. Please check your settings.");
+                    }
+                })
+            );
+        }
+
+        private static void RegisterServiceBus(
+            this IBusRegistrationConfigurator configurator,
+            AzureServiceBusOptions options,
+            BusBuilder builder)
+        {
+            configurator.UsingAzureServiceBus((provider, cfg) =>
+            {
+                var serverGroup = builder.IdOpsBuilder.Options!.ServerGroup.ToLower();
+                var environmentName = builder.IdOpsBuilder.Options!.EnvironmentName.ToLower();
+                cfg.Host(options.ConnectionString);
+                cfg.ReceiveEndpoint(
+                    $"id-{serverGroup}-{environmentName}",
+                    e =>
+                    {
+                        e.ConfigureConsumers(provider);
+                        e.PrefetchCount = options.PrefetchCount;
+                    });
+            });
+        }
+    }
+}

src/IdentityServer/Messaging.AzureServiceBus/Messaging.AzureServiceBus.csproj → src/IdentityServer/Messaging.Azure/Messaging.Azure.csproj

@@ -10,6 +10,7 @@
   <ItemGroup>
     <PackageReference Include="MassTransit.Azure.ServiceBus.Core" />
     <PackageReference Update="Microsoft.SourceLink.GitHub" />
+    <PackageReference Include="Masstransit.EventHub"/>
   </ItemGroup>
 
   <ItemGroup>

src/IdentityServer/Messaging.Azure/Options/AzureOptions.cs

@@ -0,0 +1,8 @@
+namespace IdOps.IdentityServer.Azure;
+
+public sealed class AzureOptions
+{
+    public AzureServiceBusOptions? ServiceBus { get; set; } = default!;
+    public EventHubOptions? EventHub { get; set; } = default!;
+}
+

src/IdentityServer/Messaging.Azure/Options/AzureServiceBusOptions.cs

@@ -0,0 +1,9 @@
+namespace IdOps.IdentityServer.Azure;
+
+public class AzureServiceBusOptions
+{
+    public string ConnectionString { get; set; } = default!;
+
+    public int PrefetchCount { get; set; } = 10;
+}
+

src/IdentityServer/Messaging.Azure/Options/EventHubOptions.cs

@@ -0,0 +1,8 @@
+namespace IdOps.IdentityServer.Azure;
+
+public class EventHubOptions
+{
+    public string? ConnectionString { get; set; }
+
+    public string? Namespace { get; set; }
+}

src/IdentityServer/samples/Server/IdentityServer.csproj

@@ -11,7 +11,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\..\Messaging.AzureServiceBus\Messaging.AzureServiceBus.csproj" />
+    <ProjectReference Include="..\..\Messaging.Azure\Messaging.Azure.csproj" />
     <ProjectReference Include="..\..\Messaging.RabbitMQ\Messaging.RabbitMQ.csproj" />
     <ProjectReference Include="..\..\src\IdentityServer.Core\Core.csproj" />
     <ProjectReference Include="..\..\src\Store.Mongo\Store.Mongo.csproj" />

src/IdentityServer/samples/Server/Program.cs

@@ -1,11 +1,6 @@
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
 
 namespace IdOps.IdentityServer.Samples
 {

src/IdentityServer/samples/Server/Startup.cs

@@ -1,5 +1,5 @@
 using System;
-using IdOps.IdentityServer.AzureServiceBus;
+using IdOps.IdentityServer.Azure;
 using IdOps.IdentityServer.Events;
 using IdOps.IdentityServer.RabbitMQ;
 using IdOps.IdentityServer.Samples.DataSeeding;
@@ -49,10 +49,11 @@ public void ConfigureServices(IServiceCollection services)
                         case MessagingTransport.RabbitMq:
                             busBuilder.UseRabbitMq();
                             break;
-                        case MessagingTransport.AzureServiceBus:
-                            busBuilder.UseAzureServiceBus();
+                        case MessagingTransport.Azure:
+                            busBuilder.UseAzure();
                             break;
                     }
+
                 }).AddProfileService<SampleProfileService>();
 
             services.AddSingleton<IIdOpsEventSink, ActivityEnricherSink>();

src/IdentityServer/src/Abstractions/IEventSenderWorker.cs

@@ -0,0 +1,7 @@
+using Microsoft.Extensions.Hosting;
+
+namespace IdOps.IdentityServer.Abstractions;
+
+public interface IEventSenderWorker : IHostedService
+{
+}

src/IdentityServer/src/Abstractions/IIdOpsIdentityServerBuilder.cs

@@ -6,7 +6,9 @@ namespace IdOps.IdentityServer
     public interface IIdOpsIdentityServerBuilder
     {
         IConfiguration? Configuration { get; }
+
         IServiceCollection Services { get; }
+
         IdOpsOptions Options { get; }
     }
 
@@ -30,6 +32,6 @@ public enum MessagingTransport
     {
         Memory,
         RabbitMq,
-        AzureServiceBus
+        Azure
     }
 }