diff --git a/docs/guide/messaging/transports/rabbitmq/interoperability.md b/docs/guide/messaging/transports/rabbitmq/interoperability.md index 3620df526..7f0fae84b 100644 --- a/docs/guide/messaging/transports/rabbitmq/interoperability.md +++ b/docs/guide/messaging/transports/rabbitmq/interoperability.md @@ -221,8 +221,34 @@ Wolverine = await Host.CreateDefaultBuilder().UseWolverine(opts => // Tell Wolverine to make this endpoint interoperable with MassTransit .UseMassTransitInterop(mt => { - // optionally customize the inner JSON serialization + // optionally customize the inner JSON serialization, or map the Wolverine + // tenant id from each incoming MassTransit message (see below) }) .DefaultIncomingMessage().UseForReplies(); }).StartAsync(); ``` + +### Mapping the Tenant Id + +When consuming messages from MassTransit, you can derive Wolverine's tenant id for each incoming +message from either the message body or the MassTransit envelope metadata (headers, addresses, +correlation ids). Register one or more `MapTenantIdFrom` mappers inside `UseMassTransitInterop`: + +```cs +opts.ListenToRabbitQueue("orders") + .UseMassTransitInterop(mt => + { + // Pull the tenant id straight off the strongly-typed message body + mt.MapTenantIdFrom(env => env.Message?.TenantId); + + // ...or from a MassTransit header carried on the envelope + mt.MapTenantIdFrom(env => + env.Headers.TryGetValue("tenant-id", out var value) ? value?.ToString() : null); + }); +``` + +The lambda receives the strongly-typed MassTransit envelope (`IMassTransitEnvelope`), which exposes +the deserialized `Message` alongside the MassTransit metadata. Each registration applies only to its own +message type, and returning `null` or an empty string leaves the tenant id untouched. Tenant mapping +affects only the inbound (deserialization) path, and works on any MassTransit-interop listener +(Rabbit MQ, Azure Service Bus, Amazon SQS). diff --git a/src/Testing/CoreTests/Runtime/Interop/MassTransitTenantMappingTests.cs b/src/Testing/CoreTests/Runtime/Interop/MassTransitTenantMappingTests.cs new file mode 100644 index 000000000..2c5724e91 --- /dev/null +++ b/src/Testing/CoreTests/Runtime/Interop/MassTransitTenantMappingTests.cs @@ -0,0 +1,109 @@ +using NSubstitute; +using Wolverine.Runtime.Interop.MassTransit; +using Xunit; + +namespace CoreTests.Runtime.Interop; + +public class MassTransitTenantMappingTests +{ + private readonly IMassTransitInteropEndpoint theEndpoint = Substitute.For(); + + public MassTransitTenantMappingTests() + { + // The serializer round-trips a response address through the envelope; give the fake + // endpoint a valid reply Uri so the write path doesn't emit an empty address. + theEndpoint.MassTransitReplyUri().Returns(new Uri("rabbitmq://localhost/responses")); + } + + // Round-trips a message through the MassTransit serializer: writes it in the + // MassTransit envelope format, then reads it back the way an incoming message would be + // deserialized. Returns the incoming Envelope so tests can assert on mapped metadata. + private Envelope readIncoming(MassTransitJsonSerializer serializer, object message) + { + var outgoing = new Envelope { Id = Guid.NewGuid(), Message = message }; + var data = serializer.Write(outgoing); + + var incoming = new Envelope { Data = data }; + serializer.ReadFromData(message.GetType(), incoming); + return incoming; + } + + [Fact] + public void maps_tenant_id_from_the_incoming_message_body() + { + var serializer = new MassTransitJsonSerializer(theEndpoint); + serializer.MapTenantIdFrom(env => env.Message?.TenantId); + + var incoming = readIncoming(serializer, new TenantMessage("acme", "hello")); + + incoming.TenantId.ShouldBe("acme"); + } + + [Fact] + public void maps_tenant_id_from_a_masstransit_header() + { + var serializer = new MassTransitJsonSerializer(theEndpoint); + serializer.MapTenantIdFrom(env => + env.Headers.TryGetValue("tenant-id", out var value) ? value?.ToString() : null); + + var outgoing = new Envelope { Id = Guid.NewGuid(), Message = new TenantMessage("ignored", "hello") }; + outgoing.Headers["tenant-id"] = "globex"; + var data = serializer.Write(outgoing); + + var incoming = new Envelope { Data = data }; + serializer.ReadFromData(typeof(TenantMessage), incoming); + + incoming.TenantId.ShouldBe("globex"); + } + + [Fact] + public void leaves_tenant_id_untouched_for_an_unregistered_message_type() + { + var serializer = new MassTransitJsonSerializer(theEndpoint); + serializer.MapTenantIdFrom(env => env.Message?.TenantId); + + var incoming = readIncoming(serializer, new OtherMessage("no tenant here")); + + incoming.TenantId.ShouldBeNull(); + } + + [Fact] + public void does_not_overwrite_tenant_id_when_the_source_returns_empty() + { + var serializer = new MassTransitJsonSerializer(theEndpoint); + serializer.MapTenantIdFrom(_ => ""); + + var incoming = readIncoming(serializer, new TenantMessage("acme", "hello")); + + incoming.TenantId.ShouldBeNull(); + } + + [Fact] + public void supports_multiple_registered_message_types() + { + var serializer = new MassTransitJsonSerializer(theEndpoint); + serializer + .MapTenantIdFrom(env => env.Message?.TenantId) + .MapTenantIdFrom(env => env.Message?.Org); + + readIncoming(serializer, new TenantMessage("acme", "hello")).TenantId.ShouldBe("acme"); + readIncoming(serializer, new AnotherTenantMessage("globex", "hi")).TenantId.ShouldBe("globex"); + } + + [Fact] + public void falls_through_the_whole_mapper_chain_for_an_unregistered_type() + { + var serializer = new MassTransitJsonSerializer(theEndpoint); + serializer + .MapTenantIdFrom(env => env.Message?.TenantId) + .MapTenantIdFrom(env => env.Message?.Org); + + var incoming = readIncoming(serializer, new OtherMessage("matches nothing")); + + incoming.TenantId.ShouldBeNull(); + } +} + +public record TenantMessage(string TenantId, string Body); +public record AnotherTenantMessage(string Org, string Body); +public record OtherMessage(string Body); diff --git a/src/Transports/AWS/Wolverine.AmazonSqs.Tests/masstransit_mapper_tenant_mapping.cs b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/masstransit_mapper_tenant_mapping.cs new file mode 100644 index 000000000..7644eb9d0 --- /dev/null +++ b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/masstransit_mapper_tenant_mapping.cs @@ -0,0 +1,40 @@ +using Shouldly; +using Wolverine.AmazonSqs.Internal; +using Wolverine.Runtime.Interop.MassTransit; +using Xunit; + +namespace Wolverine.AmazonSqs.Tests; + +public class masstransit_mapper_tenant_mapping +{ + // The SQS MassTransit mapper must apply the UseMassTransitInterop(configure) lambda to its + // serializer, otherwise MapTenantIdFrom (and any other serializer customization) silently + // no-ops on the Amazon SQS listener. + [Fact] + public void mapper_applies_the_configure_lambda_to_its_serializer() + { + var endpoint = new FakeMassTransitEndpoint(); + + var mapper = new MassTransitMapper(endpoint, + mt => mt.MapTenantIdFrom(env => env.Message?.TenantId)); + + var serializer = mapper.Serializer; + + var outgoing = new Envelope { Id = Guid.NewGuid(), Message = new SqsTenantMessage("acme", "hello") }; + var data = serializer.Write(outgoing); + + var incoming = new Envelope { Data = data }; + serializer.ReadFromData(typeof(SqsTenantMessage), incoming); + + incoming.TenantId.ShouldBe("acme"); + } + + private sealed class FakeMassTransitEndpoint : IMassTransitInteropEndpoint + { + public Uri? MassTransitUri() => null; + public Uri? MassTransitReplyUri() => new Uri("sqs://responses"); + public Uri? TranslateMassTransitToWolverineUri(Uri uri) => null; + } + + public record SqsTenantMessage(string TenantId, string Body); +} diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsListenerConfiguration.cs b/src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsListenerConfiguration.cs index 976685f07..27062283f 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsListenerConfiguration.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsListenerConfiguration.cs @@ -156,7 +156,7 @@ public AmazonSqsListenerConfiguration UseNServiceBusInterop(string? replyQueueNa public AmazonSqsListenerConfiguration UseMassTransitInterop(Action? configure = null) { - add(e => e.Mapper = new MassTransitMapper((Endpoint as IMassTransitInteropEndpoint)!)); + add(e => e.Mapper = new MassTransitMapper((Endpoint as IMassTransitInteropEndpoint)!, configure)); return this; } diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/MassTransitMapper.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/MassTransitMapper.cs index e0ea531b7..e8eac7344 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/MassTransitMapper.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/MassTransitMapper.cs @@ -12,10 +12,11 @@ internal class MassTransitMapper : ISqsEnvelopeMapper private readonly IMassTransitInteropEndpoint _endpoint; private MassTransitJsonSerializer _serializer; - public MassTransitMapper(IMassTransitInteropEndpoint endpoint) + public MassTransitMapper(IMassTransitInteropEndpoint endpoint, Action? configure = null) { _endpoint = endpoint; _serializer = new MassTransitJsonSerializer(endpoint); + configure?.Invoke(_serializer); } public override string ToString() => "MassTransit Interop"; diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/masstransit_interop_map_tenant_id.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/masstransit_interop_map_tenant_id.cs new file mode 100644 index 000000000..2f0a3fc36 --- /dev/null +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/masstransit_interop_map_tenant_id.cs @@ -0,0 +1,78 @@ +using System.Text; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.ComplianceTests; +using Wolverine.RabbitMQ.Internal; +using Wolverine.Runtime; +using Wolverine.Tracking; +using Wolverine.Util; +using Xunit; + +namespace Wolverine.RabbitMQ.Tests; + +public class masstransit_interop_map_tenant_id +{ + // UseMassTransitInterop(configure) must thread the configure lambda all the way down to + // the MassTransit serializer so MapTenantIdFrom takes effect on the Rabbit MQ listener. + // Uses stubbed transports (no live broker) and drives the deserialization path directly. + [Fact] + public async Task map_tenant_id_from_is_applied_on_the_rabbitmq_listener() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.DisableConventionalDiscovery(); + opts.IncludeType(); + + // Bogus host name on purpose: stubbed transports must never connect. + opts.UseRabbitMq(x => x.HostName = Guid.NewGuid().ToString()); + opts.ListenToRabbitQueue("orders") + .UseMassTransitInterop(mt => mt.MapTenantIdFrom(env => env.Message?.Tenant)); + + opts.StubAllExternalTransports(); + }).StartAsync(); + + var runtime = host.GetRuntime(); + + var endpoint = runtime.Endpoints.EndpointFor(new Uri("rabbitmq://queue/orders")) + .ShouldNotBeNull() + .ShouldBeAssignableTo(); + + // Applies UseMassTransitInterop and wires the MassTransit serializer onto the endpoint + // (a live listener does this at startup; stubbed transports skip listener startup). + endpoint.BuildMapper(runtime); + + // The wire payload a MassTransit producer sends: the real message lives under "message". + var json = $$""" + { + "messageId": "{{Guid.NewGuid()}}", + "messageType": ["urn:message:Orders:TenantOrderPlaced"], + "message": { "orderId": 92883, "tenant": "acme" } + } + """; + + var incoming = new Envelope + { + Data = Encoding.UTF8.GetBytes(json), + ContentType = "application/vnd.masstransit+json", + MessageType = typeof(TenantOrderPlaced).ToMessageTypeName(), + Destination = endpoint.Uri + }; + + await runtime.Pipeline.TryDeserializeEnvelope(incoming); + + incoming.Message.ShouldBeOfType().OrderId.ShouldBe(92883); + incoming.TenantId.ShouldBe("acme"); + } + + public record TenantOrderPlaced(int OrderId, string Tenant); + + public class TenantOrderPlacedHandler + { + // Presence registers TenantOrderPlaced in the HandlerGraph so the pipeline can + // resolve its message type during deserialization. + public void Handle(TenantOrderPlaced order) + { + } + } +} diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqEndpoint.MassTransit.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqEndpoint.MassTransit.cs index 3695f6484..37253cb16 100644 --- a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqEndpoint.MassTransit.cs +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqEndpoint.MassTransit.cs @@ -54,6 +54,6 @@ public abstract partial class RabbitMqEndpoint : IMassTransitInteropEndpoint public void UseMassTransitInterop(Action? configure = null) { - customizeMapping((m, _) => m.InteropWithMassTransit()); + customizeMapping((m, _) => m.InteropWithMassTransit(configure)); } } \ No newline at end of file diff --git a/src/Wolverine/Runtime/Interop/MassTransit/IMassTransitInterop.cs b/src/Wolverine/Runtime/Interop/MassTransit/IMassTransitInterop.cs index ed01a6a1a..a32a02653 100644 --- a/src/Wolverine/Runtime/Interop/MassTransit/IMassTransitInterop.cs +++ b/src/Wolverine/Runtime/Interop/MassTransit/IMassTransitInterop.cs @@ -10,6 +10,23 @@ public interface IMassTransitInterop /// void UseSystemTextJsonForSerialization(Action? configuration = null); + /// + /// Derive the Wolverine for incoming MassTransit messages of + /// type from the message itself or its MassTransit metadata. The + /// supplied lambda receives the strongly-typed MassTransit envelope and returns the tenant id + /// (or null / empty to leave the tenant id untouched). This only affects the inbound + /// (deserialization) path. Registering multiple message types is supported — each registration + /// applies only to its own . Registering the same type more than once + /// replaces the previous mapping for that type. + /// + /// + /// Maps the incoming MassTransit envelope to a tenant id, e.g. + /// env => env.Message?.TenantId or + /// env => env.Headers.TryGetValue("tenant-id", out var v) ? v?.ToString() : null. + /// + /// The Wolverine message type to extract the tenant id from. + IMassTransitInterop MapTenantIdFrom(Func, string?> tenantIdSource) where T : class; + // Newtonsoft.Json variant moved to WolverineFx.Newtonsoft as the // UseNewtonsoftForSerialization(this IMassTransitInterop, ...) // extension method. Install WolverineFx.Newtonsoft to opt in. diff --git a/src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs b/src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs index f8a3c8ddd..5a6db3936 100644 --- a/src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs +++ b/src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs @@ -10,7 +10,39 @@ internal interface IMassTransitEnvelope void TransferData(Envelope envelope); } -internal class MassTransitEnvelope : IMassTransitEnvelope where T : class +/// +/// Read-only, strongly-typed view over an incoming MassTransit "envelope" message. Exposed to +/// user code through hooks such as so that +/// Wolverine metadata (e.g. the tenant id) can be derived either from the deserialized message +/// body or from the surrounding MassTransit transport metadata (headers, addresses, ids). +/// +/// The Wolverine message type carried by the MassTransit envelope. +public interface IMassTransitEnvelope where T : class +{ + /// The deserialized message body. + T? Message { get; } + + string? MessageId { get; } + string? RequestId { get; } + string? CorrelationId { get; } + string? ConversationId { get; } + string? InitiatorId { get; } + + string? SourceAddress { get; } + string? ResponseAddress { get; } + string? DestinationAddress { get; } + string? FaultAddress { get; } + + string[]? MessageType { get; } + + DateTime? SentTime { get; } + DateTime? ExpirationTime { get; } + + /// The MassTransit message headers, as deserialized from the envelope. + IReadOnlyDictionary Headers { get; } +} + +internal class MassTransitEnvelope : IMassTransitEnvelope, IMassTransitEnvelope where T : class { public MassTransitEnvelope() { @@ -68,6 +100,8 @@ public MassTransitEnvelope(Envelope envelope) public Dictionary Headers { get; set; } = new(); + IReadOnlyDictionary IMassTransitEnvelope.Headers => Headers; + // Wolverine doesn't care about this, so don't bother deserializing it // ReSharper disable once UnusedMember.Global public BusHostInfo Host => BusHostInfo.Instance; diff --git a/src/Wolverine/Runtime/Interop/MassTransit/MassTransitJsonSerializer.cs b/src/Wolverine/Runtime/Interop/MassTransit/MassTransitJsonSerializer.cs index 963c90b73..3b5126f81 100644 --- a/src/Wolverine/Runtime/Interop/MassTransit/MassTransitJsonSerializer.cs +++ b/src/Wolverine/Runtime/Interop/MassTransit/MassTransitJsonSerializer.cs @@ -17,6 +17,8 @@ private IMessageSerializer private ImHashMap _uriMap = ImHashMap.Empty; + private Func? _tenantIdSource; + public MassTransitJsonSerializer(IMassTransitInteropEndpoint endpoint) { _endpoint = endpoint; @@ -37,6 +39,20 @@ public void UseSystemTextJsonForSerialization(Action? con _inner = new SystemTextJsonSerializer(options); } + public IMassTransitInterop MapTenantIdFrom(Func, string?> tenantIdSource) + where T : class + { + ArgumentNullException.ThrowIfNull(tenantIdSource); + + // Compose with any previously registered mapper so multiple message types can each + // contribute their own tenant id extraction. A mapper only fires for its own T. + var previous = _tenantIdSource; + _tenantIdSource = mtEnvelope => + mtEnvelope is IMassTransitEnvelope typed ? tenantIdSource(typed) : previous?.Invoke(mtEnvelope); + + return this; + } + /// /// Hook used by the WolverineFx.Newtonsoft package's /// UseNewtonsoftForSerialization(IMassTransitInterop) extension @@ -75,6 +91,15 @@ public object ReadFromData(Type messageType, Envelope envelope) mtEnvelope.TransferData(envelope); envelope.ReplyUri = mapResponseUri(mtEnvelope.ResponseAddress ?? mtEnvelope.SourceAddress); + if (_tenantIdSource != null) + { + var tenantId = _tenantIdSource(mtEnvelope); + if (tenantId.IsNotEmpty()) + { + envelope.TenantId = tenantId; + } + } + return mtEnvelope.Body!; }