AsyncAPI Sample

menu/menu.yaml

@@ -1727,3 +1727,5 @@
     Url: shape-the-future/redis
   - Title: NServiceBus and Infrastructure as Code (IaC)
     Url: shape-the-future/iac-provisioning
+  - Title: NServiceBus and AsyncAPI
+    Url: shape-the-future/asyncapi

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/ApiDocumentGenerator.cs

@@ -0,0 +1,88 @@
+#nullable enable
+using Json.Schema;
+using Json.Schema.Generation;
+using Microsoft.Extensions.DependencyInjection;
+using Neuroglia.AsyncApi;
+using Neuroglia.AsyncApi.FluentBuilders.v3;
+using Neuroglia.AsyncApi.Generation;
+using Neuroglia.AsyncApi.v3;
+
+namespace AsyncAPI.Feature;
+
+public class ApiDocumentGenerator(IServiceProvider serviceProvider) : IAsyncApiDocumentGenerator
+{
+    public async Task<IEnumerable<IAsyncApiDocument>> GenerateAsync(IEnumerable<Type> markupTypes, AsyncApiDocumentGenerationOptions options, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(options);
+
+        //only creating one document for the one NSB endpoint
+        var documents = new List<IAsyncApiDocument>(1);
+
+        var document = serviceProvider.GetRequiredService<IV3AsyncApiDocumentBuilder>();
+        options.V3BuilderSetup?.Invoke(document);
+
+        await GenerateChannels(document, options, cancellationToken);
+
+        documents.Add(document.Build());
+
+        return documents;
+    }
+
+    async Task GenerateChannels(IV3AsyncApiDocumentBuilder document, AsyncApiDocumentGenerationOptions options, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(document);
+        ArgumentNullException.ThrowIfNull(options);
+        IV3ChannelDefinitionBuilder channelBuilder = null!;
+
+        var typeCache = serviceProvider.GetRequiredService<TypeCache>();
+
+        //get all published events
+        foreach (var (actualType, publishedType) in typeCache.PublishedEventCache.Select(kvp => (kvp.Key, kvp.Value)))
+        {
+            var channelName = $"{publishedType.FullName!}";
+            document.WithChannel(channelName, channel =>
+            {
+                channelBuilder = channel;
+                channel
+                    .WithAddress(typeCache.EndpointName)
+                    .WithDescription(actualType.FullName);
+            });
+            await GenerateV3OperationForAsync(document, channelName, channelBuilder, actualType, publishedType, options, cancellationToken);
+        }
+
+        //NOTE this is where more channels and operations can be defined, for example subscribed to events, sent/received commands and messages
+    }
+
+    Task GenerateV3OperationForAsync(IV3AsyncApiDocumentBuilder document, string channelName, IV3ChannelDefinitionBuilder channel, Type actualType, Type producedType, AsyncApiDocumentGenerationOptions options, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(document);
+        ArgumentException.ThrowIfNullOrWhiteSpace(channelName);
+        ArgumentNullException.ThrowIfNull(channel);
+        ArgumentNullException.ThrowIfNull(actualType);
+        ArgumentNullException.ThrowIfNull(producedType);
+        ArgumentNullException.ThrowIfNull(options);
+
+        var requestMessagePayloadSchema = new JsonSchemaBuilder().FromType(actualType, JsonSchemaGeneratorConfiguration.Default).Build();
+        var messageName = producedType.FullName!;
+
+        var messageChannelReference = $"#/channels/{channelName}/messages/{producedType.FullName!}";
+        channel.WithMessage(messageName, message =>
+        {
+            message
+                .WithName(messageName)
+                .WithPayloadSchema(schema => schema
+                    .WithFormat("application/vnd.aai.asyncapi+json;version=3.0.0")
+                    .WithSchema(requestMessagePayloadSchema));
+        });
+
+        var operationName = $"{producedType.FullName!}";
+        document.WithOperation(operationName, operation =>
+        {
+            operation
+                .WithAction(V3OperationAction.Send)
+                .WithChannel($"#/channels/{channelName}")
+                .WithMessage(messageChannelReference);
+        });
+        return Task.CompletedTask;
+    }
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/AsyncAPI.Feature.csproj

@@ -0,0 +1,20 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net9.0</TargetFramework>      
+        <LangVersion>13.0</LangVersion>
+        <ImplicitUsings>enable</ImplicitUsings>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <PackageReference Include="NServiceBus" Version="9.*" />
+      <PackageReference Include="Neuroglia.AsyncApi.AspNetCore" Version="3.0.6" />
+      <PackageReference Include="Neuroglia.AsyncApi.AspNetCore.UI" Version="3.0.6" />
+      <PackageReference Include="Neuroglia.AsyncApi.Core" Version="3.0.6" />
+      <PackageReference Include="Neuroglia.AsyncApi.DependencyInjectionExtensions" Version="3.0.6" />
+      <PackageReference Include="Neuroglia.AsyncApi.FluentBuilders" Version="3.0.6" />
+      <PackageReference Include="Neuroglia.AsyncApi.Generation" Version="3.0.6" />
+      <PackageReference Include="Neuroglia.AsyncApi.Validation" Version="3.0.6" />
+    </ItemGroup>
+
+</Project>

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/AsyncApiFeature.cs

@@ -0,0 +1,95 @@
+using System.Reflection;
+using Microsoft.Extensions.DependencyInjection;
+using Neuroglia.AsyncApi.Generation;
+using NServiceBus.Features;
+using NServiceBus.Unicast.Messages;
+
+namespace AsyncAPI.Feature;
+
+public sealed class AsyncApiFeature : NServiceBus.Features.Feature
+{
+    protected override void Setup(FeatureConfigurationContext context)
+    {
+        var conventions = context.Settings.Get<Conventions>();
+
+        var messageMetadataRegistry = context.Settings.Get<MessageMetadataRegistry>();
+
+        var proxyGenerator = new TypeProxyGenerator();
+
+        Dictionary<Type, Type> publishedEventCache = new();
+        Dictionary<string, (Type SubscribedType, Type ActualType)> subscribedEventCache = new();
+
+        foreach (var messageMetadata in messageMetadataRegistry.GetAllMessages())
+        {
+            //NOTE only events decorated with the "PublishedEvent" or "SubscribedEvent" are being stored so that they can be "translated" at publish and subscribe time from their concrete types
+            if (conventions.IsEventType(messageMetadata.MessageType))
+            {
+                var publishedEvent = messageMetadata.MessageType.GetCustomAttribute<PublishedEvent>();
+                if (publishedEvent != null)
+                {
+                    publishedEventCache.Add(messageMetadata.MessageType,
+                        proxyGenerator.CreateTypeFrom($"{publishedEvent.EventName}V{publishedEvent.Version}"));
+                }
+
+                var subscribedEvent = messageMetadata.MessageType.GetCustomAttribute<SubscribedEvent>();
+                if (subscribedEvent != null)
+                {
+                    var subscribedType =
+                        proxyGenerator.CreateTypeFrom($"{subscribedEvent.EventName}V{subscribedEvent.Version}");
+                    subscribedEventCache.Add(subscribedType.FullName!,
+                        (SubscribedType: subscribedType, ActualType: messageMetadata.MessageType));
+                }
+            }
+        }
+
+        if (context.Settings.GetOrDefault<bool>("Installers.Enable"))
+        {
+            context.RegisterStartupTask(static provider => new ManualSubscribe(provider.GetRequiredService<TypeCache>()
+                .SubscribedEventCache.Values.Select(x => x.SubscribedType).ToArray()));
+        }
+
+        //Registering the behaviors required to replace the outgoing and incoming message types based on the defined conventions
+        context.Pipeline.Register(
+            static provider =>
+                new ReplaceOutgoingEnclosedMessageTypeHeaderBehavior(provider.GetRequiredService<TypeCache>().PublishedEventCache),
+            "Replaces the outgoing enclosed message type header with the published event type fullname");
+        context.Pipeline.Register(
+            static provider => new ReplaceMulticastRoutingBehavior(provider.GetRequiredService<TypeCache>().PublishedEventCache),
+            "Replaces the multicast routing strategies that match the actual published event type with the published event type name");
+
+        if (!context.Settings.GetOrDefault<bool>("Endpoint.SendOnly"))
+        {
+            context.Pipeline.Register(
+                static provider =>
+                    new ReplaceIncomingEnclosedMessageTypeHeaderBehavior(provider.GetRequiredService<TypeCache>()
+                        .SubscribedEventCache), "Replaces the incoming published event type name with the actual local event type name");
+        }
+
+        #region RegisterEventMappings
+        context.Services.AddSingleton(new TypeCache
+        {
+            EndpointName = context.Settings.EndpointName(),
+            PublishedEventCache = publishedEventCache,
+            SubscribedEventCache = subscribedEventCache
+        });
+        #endregion
+
+        #region RegisterCustomDocumentGenerator
+        context.Services.AddTransient<IAsyncApiDocumentGenerator>(
+            provider => new ApiDocumentGenerator(provider));
+        #endregion
+    }
+
+    class ManualSubscribe(Type[] subscribedEvents) : FeatureStartupTask
+    {
+        protected override Task OnStart(IMessageSession session, CancellationToken cancellationToken = default)
+        {
+            return Task.WhenAll(subscribedEvents.Select(subscribedEvent => session.Subscribe(subscribedEvent, cancellationToken: cancellationToken)));
+        }
+
+        protected override Task OnStop(IMessageSession session, CancellationToken cancellationToken = default)
+        {
+            return Task.CompletedTask;
+        }
+    }
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/EndpointConfigurationExtensions.cs

@@ -0,0 +1,19 @@
+using NServiceBus.Features;
+
+namespace AsyncAPI.Feature;
+
+public static class EndpointConfigurationExtensions
+{
+    #region EnableAsyncApiSupport
+    public static void EnableAsyncApiSupport(
+        this EndpointConfiguration endpointConfiguration)
+    {
+        endpointConfiguration.DisableFeature<AutoSubscribe>();
+        endpointConfiguration.EnableFeature<AsyncApiFeature>();
+
+        var conventions = endpointConfiguration.Conventions();
+        conventions.Add(new PublishedEventsConvention());
+        conventions.Add(new SubscribedEventsConvention());
+    }
+    #endregion
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/MessageConventions/PublishedEvent.cs

@@ -0,0 +1,8 @@
+namespace AsyncAPI.Feature;
+
+[AttributeUsage(AttributeTargets.Class, AllowMultiple = false, Inherited = false)]
+public sealed class PublishedEvent : Attribute
+{
+    public string EventName { get; init; }
+    public int Version { get; init; }
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/MessageConventions/PublishedEventsConvention.cs

@@ -0,0 +1,23 @@
+using System.Reflection;
+
+namespace AsyncAPI.Feature;
+
+public class PublishedEventsConvention : IMessageConvention
+{
+    public bool IsMessageType(Type type)
+    {
+        return false;
+    }
+
+    public bool IsCommandType(Type type)
+    {
+        return false;
+    }
+
+    public bool IsEventType(Type type)
+    {
+        return type.GetCustomAttribute<PublishedEvent>() != null;
+    }
+
+    public string Name { get; } = "AsyncAPI Sample Event Message Convention";
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/MessageConventions/ReplaceIncomingEnclosedMessageTypeHeaderBehavior.cs

@@ -0,0 +1,17 @@
+using NServiceBus.Pipeline;
+
+namespace AsyncAPI.Feature;
+
+class ReplaceIncomingEnclosedMessageTypeHeaderBehavior(Dictionary<string, (Type SubscribedType, Type ActualType)> subscribedEventCache)
+    : IBehavior<ITransportReceiveContext, ITransportReceiveContext>
+{
+    public Task Invoke(ITransportReceiveContext context, Func<ITransportReceiveContext, Task> next)
+    {
+        if (context.Message.Headers.TryGetValue(Headers.EnclosedMessageTypes, out var enclosedMessageTypes) && subscribedEventCache.TryGetValue(enclosedMessageTypes, out var subscribedEventType))
+        {
+            // very blunt and might break with certain transports
+            context.Message.Headers[Headers.EnclosedMessageTypes] = subscribedEventType.ActualType.FullName;
+        }
+        return next(context);
+    }
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/MessageConventions/ReplaceMulticastRoutingBehavior.cs

@@ -0,0 +1,39 @@
+using NServiceBus.Pipeline;
+using NServiceBus.Routing;
+
+namespace AsyncAPI.Feature;
+
+class ReplaceMulticastRoutingBehavior(Dictionary<Type, Type> publishedEventCache) : IBehavior<IRoutingContext, IRoutingContext>
+{
+    public Task Invoke(IRoutingContext context, Func<IRoutingContext, Task> next)
+    {
+        var logicalMessage = context.Extensions.Get<OutgoingLogicalMessage>();
+        if (publishedEventCache.TryGetValue(logicalMessage.MessageType, out var publishedEvent))
+        {
+            var newStrategies = new List<RoutingStrategy>(context.RoutingStrategies.Count);
+            var strategies = context.RoutingStrategies;
+            foreach (var strategy in strategies)
+            {
+                if (strategy is MulticastRoutingStrategy multicastRoutingStrategy)
+                {
+                    // we assume here a multi cast address tag will never do anything with the headers so we pass a static empty dictionary
+                    var multicastAddressTag = (MulticastAddressTag) multicastRoutingStrategy.Apply(emptyHeaders);
+                    if (multicastAddressTag.MessageType == logicalMessage.MessageType)
+                    {
+                        newStrategies.Add(new MulticastRoutingStrategy(publishedEvent));
+                    }
+                }
+                else
+                {
+                    newStrategies.Add(strategy);
+                }
+            }
+
+            context.RoutingStrategies = newStrategies;
+        }
+
+        return next(context);
+    }
+
+    private readonly Dictionary<string, string> emptyHeaders = new Dictionary<string, string>();
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/MessageConventions/ReplaceOutgoingEnclosedMessageTypeHeaderBehavior.cs

@@ -0,0 +1,19 @@
+using NServiceBus.Pipeline;
+
+namespace AsyncAPI.Feature;
+
+class ReplaceOutgoingEnclosedMessageTypeHeaderBehavior(Dictionary<Type, Type> publishedEventCache) : IBehavior<IOutgoingPhysicalMessageContext,
+    IOutgoingPhysicalMessageContext>
+{
+    public Task Invoke(IOutgoingPhysicalMessageContext context, Func<IOutgoingPhysicalMessageContext, Task> next)
+    {
+        var logicalMessage = context.Extensions.Get<OutgoingLogicalMessage>();
+        if (publishedEventCache.TryGetValue(logicalMessage.MessageType, out var publishedEvent))
+        {
+            // very blunt and might break with certain transports
+            context.Headers[Headers.EnclosedMessageTypes] = publishedEvent.FullName;
+        }
+
+        return next(context);
+    }
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/MessageConventions/SubscribedEvent.cs

@@ -0,0 +1,8 @@
+namespace AsyncAPI.Feature;
+
+[AttributeUsage(AttributeTargets.Class, AllowMultiple = false, Inherited = false)]
+public sealed class SubscribedEvent : Attribute
+{
+    public string EventName { get; init; }
+    public int Version { get; init; }
+}

samples/asyncapi/custom-message-types/Core_9/AsyncAPI.Feature/MessageConventions/SubscribedEventsConvention.cs

@@ -0,0 +1,23 @@
+using System.Reflection;
+
+namespace AsyncAPI.Feature;
+
+public class SubscribedEventsConvention : IMessageConvention
+{
+    public bool IsMessageType(Type type)
+    {
+        return false;
+    }
+
+    public bool IsCommandType(Type type)
+    {
+        return false;
+    }
+
+    public bool IsEventType(Type type)
+    {
+        return type.GetCustomAttribute<SubscribedEvent>() != null;
+    }
+
+    public string Name => "AsyncAPI Sample Event Message Convention";
+}