From 76b1d269fe49455b8e94ed2ed4350e6a770a6de0 Mon Sep 17 00:00:00 2001 From: Muhannad Al-Khatib Date: Fri, 30 Oct 2026 13:16:12 +0300 Subject: [PATCH] Publish a labelled message's contract in a "contract" header A message whose type carries SW.PrimitiveTypes.ContractAttribute is published with a "contract" header holding key@major, so a consumer can tell which shape the body has without relying on the class name. Types without the attribute, or with a malformed key, publish exactly as before: no header. The attribute is found by name, so the bus still depends on PrimitiveTypes 10.0.0 and applications that never label a message don't have to upgrade it (a raised floor would be a package-downgrade error for anyone pinning 10.0.0). Consumers only read the headers they know, so older consumers ignore it; the dead-letter copy keeps it. --- .../ContractHeaderTests.cs | 82 +++++++++++++++++++ .../SW.Bus.IntegrationTests.csproj | 4 + SW.Bus/BasicPublisher.cs | 33 +++++++- SW.Bus/BusOptions.cs | 6 ++ 4 files changed, 121 insertions(+), 4 deletions(-) create mode 100644 SW.Bus.IntegrationTests/ContractHeaderTests.cs diff --git a/SW.Bus.IntegrationTests/ContractHeaderTests.cs b/SW.Bus.IntegrationTests/ContractHeaderTests.cs new file mode 100644 index 0000000..9c1953e --- /dev/null +++ b/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"; } + +/// +/// 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. +/// +[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(); + 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 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; + } +} diff --git a/SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj b/SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj index 320c75b..f525718 100644 --- a/SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj +++ b/SW.Bus.IntegrationTests/SW.Bus.IntegrationTests.csproj @@ -13,6 +13,10 @@ + + + + diff --git a/SW.Bus/BasicPublisher.cs b/SW.Bus/BasicPublisher.cs index c75e7a7..aac5d34 100644 --- a/SW.Bus/BasicPublisher.cs +++ b/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 @@ internal class BasicPublisher( 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 Contracts = new(); + + /// + /// "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. + /// + 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 message, string exchange, byte? priority = null) { var serializerOptions = new JsonSerializerOptions() @@ -36,16 +57,16 @@ async public Task Publish(TMessage message, string exchange, byte? pri }; 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 @@ public async Task Publish(string messageTypeName, string message, string exchang } /// Written by , which also checks its size. + /// The body's contract, "key@major" (). 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 @@ public async Task Publish(string messageTypeName, byte[] message, string exchang 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); diff --git a/SW.Bus/BusOptions.cs b/SW.Bus/BusOptions.cs index b110970..14ae94c 100644 --- a/SW.Bus/BusOptions.cs +++ b/SW.Bus/BusOptions.cs @@ -12,6 +12,12 @@ public class BusOptions { private const string NodeMessageName = "__broadcast"; internal const string SourceNodeIdHeaderName = "source-node-id"; + + /// + /// 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. + /// + public const string ContractHeaderName = "contract"; const string versionPrefix = "v3."; private readonly string environment;