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;