Publish a labelled message's contract in a "contract" header

SW.Bus.IntegrationTests/ContractHeaderTests.cs

@@ -0,0 +1,82 @@
+using System;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using RabbitMQ.Client;
+using SW.PrimitiveTypes;
+using Testcontainers.RabbitMq;
+
+namespace SW.Bus.IntegrationTests;
+
+[Contract("orders.msg.placed", Major = 2)]
+public class ContractedPlaced { public string Number { get; set; } = "SO-1"; }
+
+public class UncontractedPlaced { public string Number { get; set; } = "SO-2"; }
+
+[Contract("Not A Key")]
+public class BadlyContractedPlaced { public string Number { get; set; } = "SO-3"; }
+
+/// <summary>
+/// A message whose type declares a contract carries it in the "contract" header, so a consumer can
+/// tell which shape the body has without guessing from the class name.
+/// </summary>
+[TestClass]
+public class ContractHeaderTests
+{
+    private static RabbitMqContainer _container = null!;
+
+    [ClassInitialize]
+    public static async Task ClassInitialize(TestContext _)
+    {
+        _container = Broker.Create();
+        await _container.StartAsync();
+    }
+
+    [ClassCleanup]
+    public static async Task ClassCleanup() => await _container.DisposeAsync();
+
+    [TestMethod]
+    public async Task A_contracted_message_carries_key_and_major_and_others_carry_nothing()
+    {
+        await using var harness = await BusHarness.StartAsync(_container.GetConnectionString());
+        await using var conn = await new ConnectionFactory { Uri = new Uri(_container.GetConnectionString()) }.CreateConnectionAsync();
+        await using var channel = await conn.CreateChannelAsync();
+        var queue = (await channel.QueueDeclareAsync()).QueueName;
+        foreach (var type in new[] { nameof(ContractedPlaced), nameof(UncontractedPlaced), nameof(BadlyContractedPlaced) })
+            await channel.QueueBindAsync(queue, harness.Options.ProcessExchange, type.ToLower());
+
+        await using (var scope = harness.Services.CreateAsyncScope())
+        {
+            var publish = scope.ServiceProvider.GetRequiredService<IPublish>();
+            await publish.Publish(new ContractedPlaced());
+            await publish.Publish(new UncontractedPlaced());
+            await publish.Publish(new BadlyContractedPlaced());
+        }
+
+        Assert.AreEqual("orders.msg.placed@2", await ContractHeaderOf(channel, queue, "SO-1"));
+        Assert.IsNull(await ContractHeaderOf(channel, queue, "SO-2"), "a type with no contract gets no header");
+        Assert.IsNull(await ContractHeaderOf(channel, queue, "SO-3"), "a malformed key is left out, not published");
+    }
+
+    private static async Task<string?> ContractHeaderOf(IChannel channel, string queue, string number)
+    {
+        for (var attempt = 0; attempt < 50; attempt++)
+        {
+            var message = await channel.BasicGetAsync(queue, autoAck: false);
+            if (message == null) { await Task.Delay(100); continue; }
+            if (!Encoding.UTF8.GetString(message.Body.Span).Contains(number))
+            {
+                await channel.BasicNackAsync(message.DeliveryTag, false, requeue: true);
+                continue;
+            }
+            await channel.BasicAckAsync(message.DeliveryTag, false);
+            return message.BasicProperties.Headers != null
+                   && message.BasicProperties.Headers.TryGetValue(BusOptions.ContractHeaderName, out var value)
+                ? Encoding.UTF8.GetString((byte[])value!)
+                : null;
+        }
+        Assert.Fail($"The message with {number} never arrived.");
+        return null;
+    }
+}

SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj

@@ -13,6 +13,10 @@
     <PackageReference Include="Testcontainers.RabbitMq" Version="3.10.0" />
   </ItemGroup>
 
+  <ItemGroup>
+    <!-- ContractAttribute: the library itself stays on 10.0.0 and finds the attribute by name. -->
+    <PackageReference Include="SimplyWorks.PrimitiveTypes" Version="10.0.2" />
+  </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\SW.Bus\SW.Bus.csproj" />
     <ProjectReference Include="..\SW.Bus.RabbitMqExtensions\SW.Bus.RabbitMqExtensions.csproj" />

SW.Bus/BasicPublisher.cs

@@ -1,6 +1,8 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Linq;
 using System.Security.Claims;
 using System.Text;
 using System.Text.Json;
@@ -28,6 +30,25 @@
     private readonly SemaphoreSlim channelLock = new(1, 1);
     private IChannel channel;
 
+    // A type's attributes never change, so each type is looked up once per process.
+    private static readonly ConcurrentDictionary<Type, string> Contracts = new();
+
+    /// <summary>
+    /// "key@major" when the type declares a well-formed SW.PrimitiveTypes.ContractAttribute, else null.
+    /// Found by name, not by type: the attribute exists from PrimitiveTypes 10.0.2, and the bus still
+    /// works with 10.0.0, so applications that never label a message don't have to upgrade.
+    /// </summary>
+    internal static string ContractOf(Type type) => Contracts.GetOrAdd(type, t =>
+    {
+        var attribute = t.GetCustomAttributes(inherit: false)
+            .FirstOrDefault(a => a.GetType().FullName == "SW.PrimitiveTypes.ContractAttribute");
+        if (attribute == null) return null;
+        var attributeType = attribute.GetType();
+        return attributeType.GetProperty("IsWellFormed")?.GetValue(attribute) is true
+            ? attributeType.GetProperty("Id")?.GetValue(attribute) as string
+            : null;
+    });
+
     async public Task Publish<TMessage>(TMessage message, string exchange, byte? priority = null)
     {
         var serializerOptions = new JsonSerializerOptions()
@@ -36,16 +57,16 @@
         };
         var body = JsonSerializer.Serialize(message,message.GetType(), serializerOptions);
 
-        await Publish(message.GetType().Name, body,exchange, priority);
+        await Publish(message.GetType().Name, body,exchange, priority, contract: ContractOf(message.GetType()));
     }
 
     public async Task Publish(string messageTypeName, string message, string exchange, byte? priority = null,
-        string valuesHeader = null)
+        string valuesHeader = null, string contract = null)
     {
         try
         {
             var body = Encoding.UTF8.GetBytes(message);
-            await Publish(messageTypeName, body, exchange, priority, valuesHeader);
+            await Publish(messageTypeName, body, exchange, priority, valuesHeader, contract);
         }
         catch (Exception e)
         {
@@ -55,8 +76,9 @@
     }
 
     /// <param name="valuesHeader">Written by <see cref="PublishedValues.ToHeader"/>, which also checks its size.</param>
+    /// <param name="contract">The body's contract, "key@major" (<see cref="BusOptions.ContractHeaderName"/>).</param>
     public async Task Publish(string messageTypeName, byte[] message, string exchange, byte? priority = null,
-        string valuesHeader = null)
+        string valuesHeader = null, string contract = null)
     {
         var activity = BusDiagnostics.ActivitySource.StartActivity($"bus.publish {messageTypeName}", ActivityKind.Producer);
         var stopwatch = Stopwatch.StartNew();
@@ -101,6 +123,9 @@
         if (valuesHeader != null)
             props.Headers[RequestContext.ValuesHeaderName] = valuesHeader;
 
+        if (contract != null)
+            props.Headers[BusOptions.ContractHeaderName] = contract;
+
         props.Headers.Add(BusOptions.SourceNodeIdHeaderName,busOptions.NodeId);
         props.Headers.Add("Id", props.MessageId);
 

SW.Bus/BusOptions.cs

@@ -12,6 +12,12 @@ public class BusOptions
     {
         private const string NodeMessageName = "__broadcast";
         internal const string SourceNodeIdHeaderName = "source-node-id";
+
+        /// <summary>
+        /// The header carrying the published type's contract, "key@major", when the type declares one with
+        /// SW.PrimitiveTypes.ContractAttribute (PrimitiveTypes 10.0.2 and later). Absent otherwise.
+        /// </summary>
+        public const string ContractHeaderName = "contract";
         const string versionPrefix = "v3.";
         private readonly string environment;