diff --git a/src/Mocha/src/Mocha/Consumers/Extensions/ConsumeContextExtensions.cs b/src/Mocha/src/Mocha/Consumers/Extensions/ConsumeContextExtensions.cs index dbd65a2fa3c..19919134be5 100644 --- a/src/Mocha/src/Mocha/Consumers/Extensions/ConsumeContextExtensions.cs +++ b/src/Mocha/src/Mocha/Consumers/Extensions/ConsumeContextExtensions.cs @@ -11,8 +11,9 @@ internal static class ConsumeContextExtensions /// Creates reply options from the incoming message metadata when a response channel is available. /// /// - /// Correlation id and headers are copied so replies/faults remain linked to the original request - /// and downstream workflows (for example saga headers) keep working. + /// Headers are copied so replies and faults remain linked to the original request and downstream + /// workflows (for example saga headers) keep working. The correlation id is echoed when present, + /// so callers that correlate by a different mechanism (such as a saga header) are still supported. /// public static bool TryCreateResponseOptions(this IConsumeContext context, out ReplyOptions options) { @@ -23,15 +24,10 @@ public static bool TryCreateResponseOptions(this IConsumeContext context, out Re return false; } - if (context.CorrelationId is not { } correlationId) - { - return false; - } - options = new ReplyOptions { Headers = [], - CorrelationId = correlationId, + CorrelationId = context.CorrelationId, ConversationId = context.ConversationId, ReplyAddress = replyTo }; diff --git a/src/Mocha/src/Mocha/Descriptions/InboundRouteDescription.cs b/src/Mocha/src/Mocha/Descriptions/InboundRouteDescription.cs index c88dc61514e..06726918a4b 100644 --- a/src/Mocha/src/Mocha/Descriptions/InboundRouteDescription.cs +++ b/src/Mocha/src/Mocha/Descriptions/InboundRouteDescription.cs @@ -3,12 +3,14 @@ namespace Mocha; /// /// Describes an inbound route binding a message type to a consumer and endpoint. /// -/// The kind of inbound route (subscribe, send, or request). +/// The kind of inbound route (subscribe, send, request, or reply). /// The identity string of the message type, or null if unknown. /// The name of the consumer handling messages on this route, or null if unbound. +/// The condition that decides whether this route selects its consumer for a received message. /// The endpoint reference, or null if not yet assigned. public sealed record InboundRouteDescription( InboundRouteKind Kind, string? MessageTypeIdentity, string? ConsumerName, + RouteConditionDescription Condition, EndpointReferenceDescription? Endpoint); diff --git a/src/Mocha/src/Mocha/Descriptions/RouteConditionDescription.cs b/src/Mocha/src/Mocha/Descriptions/RouteConditionDescription.cs new file mode 100644 index 00000000000..848b7c1b253 --- /dev/null +++ b/src/Mocha/src/Mocha/Descriptions/RouteConditionDescription.cs @@ -0,0 +1,15 @@ +namespace Mocha; + +/// +/// Describes the match condition that determines whether an inbound route selects its consumer for a +/// received message, for diagnostic and visualization purposes. +/// +/// The kind of condition, such as the message type rule, a header rule, or a composite. +/// +/// A condition specific detail, such as a message type identity or a header key, or null if not applicable. +/// +/// The nested conditions of a composite condition, or an empty list for a leaf condition. +public sealed record RouteConditionDescription( + string Kind, + string? Detail, + IReadOnlyList Children); diff --git a/src/Mocha/src/Mocha/MessageTypes/Conditions/AndCondition.cs b/src/Mocha/src/Mocha/MessageTypes/Conditions/AndCondition.cs new file mode 100644 index 00000000000..bf304422d75 --- /dev/null +++ b/src/Mocha/src/Mocha/MessageTypes/Conditions/AndCondition.cs @@ -0,0 +1,43 @@ +using System.Collections.Immutable; +using Mocha.Middlewares; + +namespace Mocha; + +/// +/// Matches only when all of its child conditions match. +/// +/// +/// Initializes a new instance of the class. +/// +/// The child conditions that must all match. +internal sealed class AndCondition(ImmutableArray conditions) : RouteCondition +{ + /// + public override void Initialize(IMessagingConfigurationContext context) + { + foreach (var condition in conditions) + { + condition.Initialize(context); + } + } + + /// + public override bool Matches(IReceiveContext context) + { + foreach (var condition in conditions) + { + if (!condition.Matches(context)) + { + return false; + } + } + + return true; + } + + /// + public override RouteConditionDescription Describe() + => new("And", null, [.. conditions.Select(static c => c.Describe())]); + + public static AndCondition Create(params ReadOnlySpan conditions) => new([.. conditions]); +} diff --git a/src/Mocha/src/Mocha/MessageTypes/Conditions/HeaderPresentCondition.cs b/src/Mocha/src/Mocha/MessageTypes/Conditions/HeaderPresentCondition.cs new file mode 100644 index 00000000000..1eed87b861d --- /dev/null +++ b/src/Mocha/src/Mocha/MessageTypes/Conditions/HeaderPresentCondition.cs @@ -0,0 +1,19 @@ +using Mocha.Middlewares; + +namespace Mocha; + +/// +/// Matches when the received message carries a header for a given key, regardless of its value. +/// +/// The typed key of the header that must be present. +/// The type of the header value. +internal sealed class HeaderPresentCondition(ContextDataKey key) : RouteCondition +{ + /// + public override bool Matches(IReceiveContext context) + => context.Headers.TryGet(key, out _); + + /// + public override RouteConditionDescription Describe() + => new("HeaderPresent", key.Key, []); +} diff --git a/src/Mocha/src/Mocha/MessageTypes/Conditions/MessageTypeCondition.cs b/src/Mocha/src/Mocha/MessageTypes/Conditions/MessageTypeCondition.cs new file mode 100644 index 00000000000..6c1213acb61 --- /dev/null +++ b/src/Mocha/src/Mocha/MessageTypes/Conditions/MessageTypeCondition.cs @@ -0,0 +1,67 @@ +using Mocha.Middlewares; + +namespace Mocha; + +/// +/// Matches when the received message type is the route's message type or encloses it through its type +/// hierarchy, so handlers registered for base contracts can receive derived messages. When the route +/// is optional and the received message has no resolved message type, the condition still matches so +/// the route can select on other terms. +/// +internal sealed class MessageTypeCondition : RouteCondition +{ + private readonly Type _eventType; + private readonly bool _optional; + private MessageType? _messageType; + + /// + /// Initializes a new instance of the class for the given CLR + /// message type. The message type is resolved against the registry in . + /// + /// The CLR type of the message the route handles. + /// + /// When true, the condition matches a received message that has no resolved message type. + /// + public MessageTypeCondition(Type eventType, bool optional = false) + { + _eventType = eventType; + _optional = optional; + } + + /// + /// Initializes a new instance of the class from an already + /// resolved message type. + /// + /// The resolved message type the route handles. + /// + /// When true, the condition matches a received message that has no resolved message type. + /// + public MessageTypeCondition(MessageType messageType, bool optional = false) + { + _eventType = messageType.RuntimeType; + _messageType = messageType; + _optional = optional; + } + + public MessageType? MessageType => _messageType; + + /// + public override void Initialize(IMessagingConfigurationContext context) + => _messageType ??= context.Messages.GetOrAdd(context, _eventType); + + /// + public override bool Matches(IReceiveContext context) + { + if (context.MessageType is not { } mt) + { + return _optional; + } + + return _messageType is { } messageType + && (mt == messageType || mt.EnclosedMessageTypes.Contains(messageType)); + } + + /// + public override RouteConditionDescription Describe() + => new("MessageType", _messageType?.Identity, []); +} diff --git a/src/Mocha/src/Mocha/MessageTypes/Conditions/NoMatchCondition.cs b/src/Mocha/src/Mocha/MessageTypes/Conditions/NoMatchCondition.cs new file mode 100644 index 00000000000..ff91332aeb6 --- /dev/null +++ b/src/Mocha/src/Mocha/MessageTypes/Conditions/NoMatchCondition.cs @@ -0,0 +1,26 @@ +using Mocha.Middlewares; + +namespace Mocha; + +/// +/// Never matches. Assigned to the reply route that has no message type so the generic receive router +/// has a non-null condition to evaluate while keeping the route inert. +/// +internal sealed class NoMatchCondition : RouteCondition +{ + private NoMatchCondition() + { + } + + /// + public override bool Matches(IReceiveContext context) => false; + + /// + public override RouteConditionDescription Describe() + => new("NoMatch", null, []); + + /// + /// Gets the shared instance of the . + /// + public static NoMatchCondition Instance { get; } = new(); +} diff --git a/src/Mocha/src/Mocha/MessageTypes/Conditions/RouteCondition.cs b/src/Mocha/src/Mocha/MessageTypes/Conditions/RouteCondition.cs new file mode 100644 index 00000000000..248b377a490 --- /dev/null +++ b/src/Mocha/src/Mocha/MessageTypes/Conditions/RouteCondition.cs @@ -0,0 +1,33 @@ +using Mocha.Middlewares; + +namespace Mocha; + +/// +/// Represents the rule that decides whether an inbound route selects its consumer for a received +/// message. The receive router evaluates the condition against the message envelope metadata, the +/// message type and the headers, both of which are available before the message body is deserialized. +/// +public abstract class RouteCondition +{ + /// + /// Prepares the condition against the messaging configuration so that any message types it + /// references are resolved and registered before the route starts evaluating received messages. + /// + /// The messaging configuration context. + public virtual void Initialize(IMessagingConfigurationContext context) + { + } + + /// + /// Determines whether the route should select its consumer for the given received message. + /// + /// The receive context exposing the resolved message type and the headers. + /// true if the route matches the message; otherwise, false. + public abstract bool Matches(IReceiveContext context); + + /// + /// Creates a structured description of this condition for visualization and diagnostic purposes. + /// + /// A representing this condition. + public abstract RouteConditionDescription Describe(); +} diff --git a/src/Mocha/src/Mocha/MessageTypes/Configurations/InboundRouteConfiguration.cs b/src/Mocha/src/Mocha/MessageTypes/Configurations/InboundRouteConfiguration.cs index 9329d1e0256..a10f41548ec 100644 --- a/src/Mocha/src/Mocha/MessageTypes/Configurations/InboundRouteConfiguration.cs +++ b/src/Mocha/src/Mocha/MessageTypes/Configurations/InboundRouteConfiguration.cs @@ -29,4 +29,10 @@ public class InboundRouteConfiguration : MessagingConfiguration /// Gets or sets the kind of inbound route. /// public InboundRouteKind Kind { get; set; } + + /// + /// Gets or sets the condition that decides whether this route selects its consumer for a received + /// message, or null to derive the default condition from the message type. + /// + public RouteCondition? Condition { get; set; } } diff --git a/src/Mocha/src/Mocha/MessageTypes/Descriptors/IInboundRouteDescriptor.cs b/src/Mocha/src/Mocha/MessageTypes/Descriptors/IInboundRouteDescriptor.cs index adaf2e0b56a..05c7ed2fddd 100644 --- a/src/Mocha/src/Mocha/MessageTypes/Descriptors/IInboundRouteDescriptor.cs +++ b/src/Mocha/src/Mocha/MessageTypes/Descriptors/IInboundRouteDescriptor.cs @@ -25,4 +25,12 @@ public interface IInboundRouteDescriptor : IMessagingDescriptorThe inbound route kind. /// This descriptor for method chaining. IInboundRouteDescriptor Kind(InboundRouteKind kind); + + /// + /// Sets the condition that decides whether this route selects its consumer for a received message, + /// overriding the default condition derived from the message type. + /// + /// The route condition. + /// This descriptor for method chaining. + IInboundRouteDescriptor Condition(RouteCondition condition); } diff --git a/src/Mocha/src/Mocha/MessageTypes/Descriptors/InboundRouteDescriptor.cs b/src/Mocha/src/Mocha/MessageTypes/Descriptors/InboundRouteDescriptor.cs index 110e1c0a367..de5aabcb6b9 100644 --- a/src/Mocha/src/Mocha/MessageTypes/Descriptors/InboundRouteDescriptor.cs +++ b/src/Mocha/src/Mocha/MessageTypes/Descriptors/InboundRouteDescriptor.cs @@ -38,6 +38,13 @@ public IInboundRouteDescriptor Kind(InboundRouteKind kind) return this; } + /// + public IInboundRouteDescriptor Condition(RouteCondition condition) + { + Configuration.Condition = condition; + return this; + } + /// /// Creates the final configuration from the descriptor state. /// diff --git a/src/Mocha/src/Mocha/MessageTypes/InboundRoute.cs b/src/Mocha/src/Mocha/MessageTypes/InboundRoute.cs index 1c6de1b4dba..6d5111ed727 100644 --- a/src/Mocha/src/Mocha/MessageTypes/InboundRoute.cs +++ b/src/Mocha/src/Mocha/MessageTypes/InboundRoute.cs @@ -34,6 +34,11 @@ public sealed class InboundRoute /// public InboundRouteKind Kind { get; private set; } + /// + /// Gets the condition that decides whether this route selects its consumer for a received message. + /// + public RouteCondition Condition { get; private set; } = null!; + /// /// Gets the receive endpoint that this route is connected to, or null if not yet connected. /// @@ -71,6 +76,13 @@ public void Initialize(IMessagingConfigurationContext context, InboundRouteConfi context.Messages.GetOrAdd(context, configuration.ResponseRuntimeType); } + Condition = configuration.Condition + ?? (MessageType is not null + ? new MessageTypeCondition(MessageType) + : NoMatchCondition.Instance); + + Condition.Initialize(context); + MarkInitialized(); } @@ -146,6 +158,7 @@ public InboundRouteDescription Describe() Kind, MessageType?.Identity, Consumer?.Name, + Condition.Describe(), Endpoint is not null ? new EndpointReferenceDescription(Endpoint.Name, Endpoint.Address?.ToString(), Endpoint.Transport.Name) : null); diff --git a/src/Mocha/src/Mocha/Middlewares/DefaultMessageBus.cs b/src/Mocha/src/Mocha/Middlewares/DefaultMessageBus.cs index c3996080adb..2bb6363b01b 100644 --- a/src/Mocha/src/Mocha/Middlewares/DefaultMessageBus.cs +++ b/src/Mocha/src/Mocha/Middlewares/DefaultMessageBus.cs @@ -429,6 +429,10 @@ public async ValueTask CancelScheduledMessageAsync(string token, Cancellat private void PropagateCorrelationIds(DispatchContext context) { + // ConversationId and CausationId are lineage ids that flow through the whole graph. + // CorrelationId is a per-hop routing key (request/reply pairing, saga correlation) and must + // be set explicitly, never inherited, or an outbound message could complete the wrong + // request promise or attach to the wrong saga. if (consumeContextAccessor.Context is { } ambient) { context.ConversationId ??= ambient.ConversationId; diff --git a/src/Mocha/src/Mocha/Middlewares/Receive/RoutingMiddleware.cs b/src/Mocha/src/Mocha/Middlewares/Receive/RoutingMiddleware.cs index fd01acc2600..2786a42a145 100644 --- a/src/Mocha/src/Mocha/Middlewares/Receive/RoutingMiddleware.cs +++ b/src/Mocha/src/Mocha/Middlewares/Receive/RoutingMiddleware.cs @@ -4,11 +4,13 @@ namespace Mocha.Middlewares; /// -/// Selects matching consumers for the resolved message type and current endpoint. +/// Selects matching consumers for the current endpoint by evaluating each route's condition against +/// the received message. /// /// -/// Matches include enclosed message types so handlers registered for base contracts can receive -/// derived messages. +/// The default condition matches by message type, including enclosed message types so handlers +/// registered for base contracts can receive derived messages. Other conditions, such as header based +/// reply routing, select on envelope metadata alone. /// Without this middleware, no consumer list is built for execution and messages can traverse the /// pipeline without ever reaching application handlers. /// @@ -18,21 +20,12 @@ public async ValueTask InvokeAsync(IReceiveContext context, ReceiveDelegate next { var feature = context.Features.GetOrSet(); - if (context.MessageType is { } messageType) + foreach (var route in router.GetInboundByEndpoint(context.Endpoint)) { - var routes = router.GetInboundByEndpoint(context.Endpoint); - - foreach (var route in routes) + if (route.Consumer is not null && route.Condition.Matches(context)) { - if (route.MessageType is not null - && route.Consumer is not null - && ( - route.MessageType == messageType - || messageType.EnclosedMessageTypes.Contains(route.MessageType))) - { - // Consumers are collected on the feature for later execution middleware. - feature.Consumers.Add(route.Consumer); - } + // Consumers are collected on the feature for later execution middleware. + feature.Consumers.Add(route.Consumer); } } diff --git a/src/Mocha/src/Mocha/Sagas/Saga.cs b/src/Mocha/src/Mocha/Sagas/Saga.cs index 69ad93600d6..bd6bc9c6f7a 100644 --- a/src/Mocha/src/Mocha/Sagas/Saga.cs +++ b/src/Mocha/src/Mocha/Sagas/Saga.cs @@ -566,6 +566,9 @@ private async Task SendEventsAsync( var requestType = context.Runtime.GetMessageType(message.GetType()); var endpoint = context.Runtime.GetSendEndpoint(requestType); + // Route the reply to the shared reply endpoint, where the saga's OnReply/OnAnyReply route + // is bound. The reply is delivered to the saga consumer there and correlated by the saga + // header, so no correlation id is required. options = options with { ReplyEndpoint = endpoint.Transport.ReplyReceiveEndpoint?.Source.Address }; options.Headers.Set(SagaContextData.SagaId, state.Id.ToString("D")); diff --git a/src/Mocha/src/Mocha/Sagas/SagaEventListener.cs b/src/Mocha/src/Mocha/Sagas/SagaEventListener.cs index 9aaacdf6d95..7c38237ae7a 100644 --- a/src/Mocha/src/Mocha/Sagas/SagaEventListener.cs +++ b/src/Mocha/src/Mocha/Sagas/SagaEventListener.cs @@ -18,23 +18,53 @@ protected override void Configure(IConsumerDescriptor descriptor) { foreach (var transition in state.Value.Transitions) { + var eventType = transition.Key; + var transitionKind = transition.Value.TransitionKind; + descriptor.AddRoute(r => - r.MessageType(transition.Key) + { + r.MessageType(eventType) .Kind( - transition.Value.TransitionKind switch + transitionKind switch { SagaTransitionKind.Event => InboundRouteKind.Subscribe, SagaTransitionKind.Send => InboundRouteKind.Send, SagaTransitionKind.Request => InboundRouteKind.Request, SagaTransitionKind.Reply => InboundRouteKind.Reply, - _ => throw new InvalidOperationException( - $"Invalid transition kind: {transition.Value.TransitionKind}") - }) - ); + _ => throw new InvalidOperationException($"Invalid transition kind: {transitionKind}") + }); + + if (transitionKind == SagaTransitionKind.Reply) + { + // A reply lands on the shared reply endpoint alongside non saga (RPC) replies. + // The saga-id header marks replies to the saga's own requests, so the route + // selects only those. A typed reply additionally narrows by its message type. + r.Condition(CreateReplyCondition(eventType)); + } + }); } } } + private static RouteCondition CreateReplyCondition(Type eventType) + { + // The saga-id header is the discriminator: only replies to the saga's own requests carry it, + // so the route never selects a non saga (RPC) reply on the shared reply endpoint. + var sagaId = new HeaderPresentCondition(SagaContextData.SagaId); + + // OnAnyReply (OnReply) routes every saga-id reply from the endpoint to the saga + // consumer, which then correlates by id. + if (eventType == typeof(object)) + { + return sagaId; + } + + // A typed OnReply requires the saga-id and, when the received message resolves a message + // type, requires it to match the reply type. A reply with no resolved message type still + // selects the route on the saga-id alone. + return AndCondition.Create(sagaId, new MessageTypeCondition(eventType, optional: true)); + } + /// public override ConsumerDescription Describe() { diff --git a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs index d2746abb021..e80a8c57028 100644 --- a/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs +++ b/src/Mocha/test/Mocha.Sagas.Tests/IntegrationTests.cs @@ -133,7 +133,7 @@ public async Task Saga_Should_TimeoutWithCustomResponse() Assert.Equal(0, storage.Count); } - [Fact(Skip = "OnAnyReply saga reply routing requires full reply pipeline investigation")] + [Fact] public async Task Saga_Should_SupportRequestResponse() { // arrange @@ -158,6 +158,175 @@ public async Task Saga_Should_SupportRequestResponse() Assert.Equal(0, storage.Count); } + [Fact] + public async Task Saga_Should_ReceiveReply_When_SendUsedWithOnReply() + { + // A saga that uses .Send to dispatch a request and .OnReply (or .OnAnyReply) to handle the + // response routes the reply back to its own durable endpoint and correlates it by the saga + // header, even though the reply type does not match any subscribed route. This test isolates + // the reply leg: the handler runs, the reply is routed to the saga, and the saga finalizes. + + // arrange + var recorder = new MessageRecorder(); + await using var provider = await CreateBusAsync(b => + { + b.Services.AddSingleton(recorder); + b.AddRequestHandler(); + b.AddSaga(); + }); + + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + var storage = provider.GetRequiredService(); + + // act - publish StartRequestEvent to start the saga, which sends TriggerRequest via .Send + await bus.PublishAsync(new StartRequestEvent(), CancellationToken.None); + + // assert - the handler runs, proving the request was delivered (this isolates the reply leg) + Assert.True(await recorder.WaitAsync(s_timeout), "request handler never executed"); + + // give the reply time to route back to the saga and finalize it + var deadline = DateTime.UtcNow + s_timeout; + while (storage.Count != 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(50, default); + } + + // assert - the reply routed back to the saga, transitioned it to its final state, and the + // saga state was deleted from the store (storage.Count reaches 0). + Assert.True( + storage.Count == 0, + "the request handler executed but the reply was not routed back to the saga: the saga " + + "did not reach its final state and its state was not deleted from the store."); + } + + [Fact] + public async Task Saga_Should_ReceiveReply_When_SendUsedWithTypedOnReply() + { + // A saga that uses .Send and a typed .OnReply() (not the object based + // OnAnyReply) must finalize. This proves a typed reply resolves its concrete type on the + // shared reply endpoint and selects the typed reply route. + + // arrange + var recorder = new MessageRecorder(); + await using var provider = await CreateBusAsync(b => + { + b.Services.AddSingleton(recorder); + b.AddRequestHandler(); + b.AddSaga(); + }); + + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + var storage = provider.GetRequiredService(); + + // act - publish StartRequestEvent to start the saga, which sends TriggerRequest via .Send + await bus.PublishAsync(new StartRequestEvent(), CancellationToken.None); + + // assert - the handler runs, proving the request was delivered + Assert.True(await recorder.WaitAsync(s_timeout), "request handler never executed"); + + // give the typed reply time to route back to the saga and finalize it + var deadline = DateTime.UtcNow + s_timeout; + while (storage.Count != 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(50, default); + } + + // assert - the typed reply routed back to the saga and finalized it + Assert.True( + storage.Count == 0, + "the request handler executed but the typed reply was not routed back to the saga: the " + + "saga did not reach its final state and its state was not deleted from the store."); + } + + [Fact] + public async Task Sagas_Should_EachOwnTheirReply_When_TwoSagasUseOnAnyReply() + { + // Two sagas both declare OnAnyReply, so a saga-id reply selects both saga consumers. The + // condition selects the consumer, never the instance. The saga that owns the reply's saga-id + // transitions and finalizes; the other loads no instance for that id and no-ops, with no + // phantom create. Both sagas reach their final state and the store ends empty. + + // arrange + await using var provider = await CreateBusAsync(b => + { + b.AddRequestHandler(); + b.AddRequestHandler(); + b.AddSaga(); + b.AddSaga(); + }); + + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + var storage = provider.GetRequiredService(); + + // act - start both sagas; each sends its own request and awaits its own reply + await bus.PublishAsync(new StartRequestEvent(), CancellationToken.None); + await bus.PublishAsync(new StartSecondRequestEvent(), CancellationToken.None); + + var deadline = DateTime.UtcNow + s_timeout; + while (storage.Count != 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(50, default); + } + + // assert - both sagas finalized and were removed, with no phantom instances left behind + Assert.Equal(0, storage.Count); + } + + [Fact] + public async Task RpcReply_Should_NotSelectSaga_When_SameResponseTypeRequestedDirectly() + { + // RPC-contamination regression: a saga declares OnReply and the same process + // issues a direct bus.RequestAsync of the same response type. The RPC reply carries no saga-id + // header, so the saga reply route's saga-id gate excludes it: the RPC round-trips and the saga + // is never selected (which would otherwise fault via CreateState). + + // arrange + await using var provider = await CreateBusAsync(b => + { + b.AddRequestHandler(); + b.AddSaga(); + }); + + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + var storage = provider.GetRequiredService(); + + // act - a direct RPC request of the saga's reply type, with no saga running + var response = await bus.RequestAsync(new TriggerRequest(), CancellationToken.None); + + // assert - the RPC round-trips and no saga state was created from the RPC reply + Assert.IsType(response); + Assert.Equal(0, storage.Count); + } + + [Fact] + public async Task Bus_Should_RoundTripReply_When_RequestAsyncUsedDirectly() + { + // baseline: a direct bus.RequestAsync sets a CorrelationId and registers a deferred response + // promise, so the reply round-trips. This contrasts the RPC path with the saga .Send path. + + // arrange + var recorder = new MessageRecorder(); + await using var provider = await CreateBusAsync(b => + { + b.Services.AddSingleton(recorder); + b.AddRequestHandler(); + }); + + using var scope = provider.CreateScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + + // act + var response = await bus.RequestAsync(new TriggerRequest(), CancellationToken.None); + + // assert + Assert.NotNull(response); + Assert.IsType(response); + } + // ========================================================================= // Saga Definitions // ========================================================================= @@ -258,6 +427,49 @@ protected override void Configure(ISagaDescriptor descript } } + /// + /// Second request-response saga: StartSecondRequestEvent -> AwaitingResponse (sends + /// SecondTriggerRequest) -> any reply -> Completed (final). Used to prove OnAnyReply fan-out is + /// safe across two sagas. + /// + public sealed class SecondRequestResponseSaga : Saga + { + protected override void Configure(ISagaDescriptor descriptor) + { + descriptor + .Initially() + .OnEvent() + .StateFactory(_ => new RequestResponseState()) + .Send((_, _) => new SecondTriggerRequest()) + .TransitionTo("AwaitingResponse"); + + descriptor.During("AwaitingResponse").OnAnyReply().TransitionTo("Completed"); + + descriptor.Finally("Completed"); + } + } + + /// + /// Typed-reply saga: StartRequestEvent -> AwaitingResponse (sends TriggerRequest) -> + /// typed OnReply<TriggerResponse> -> Completed (final) + /// + public sealed class TypedReplySaga : Saga + { + protected override void Configure(ISagaDescriptor descriptor) + { + descriptor + .Initially() + .OnEvent() + .StateFactory(_ => new RequestResponseState()) + .Send((_, _) => new TriggerRequest()) + .TransitionTo("AwaitingResponse"); + + descriptor.During("AwaitingResponse").OnReply().TransitionTo("Completed"); + + descriptor.Finally("Completed"); + } + } + // ========================================================================= // Events & Messages // ========================================================================= @@ -270,6 +482,8 @@ public sealed class EndTimeoutEvent; public sealed class StartRequestEvent; + public sealed class StartSecondRequestEvent; + public sealed record TriggerEvent(Guid? CorrelationId) : ICorrelatable; public sealed record TestMessage(Guid Id); @@ -278,6 +492,10 @@ public sealed record TriggerRequest : IEventRequest; public sealed record TriggerResponse; + public sealed record SecondTriggerRequest : IEventRequest; + + public sealed record SecondTriggerResponse; + // ========================================================================= // Handlers // ========================================================================= @@ -308,6 +526,27 @@ public ValueTask HandleAsync(TriggerRequest request, Cancellati } } + private sealed class SecondTriggerRequestHandler + : IEventRequestHandler + { + public ValueTask HandleAsync( + SecondTriggerRequest request, + CancellationToken cancellationToken) + { + return new(new SecondTriggerResponse()); + } + } + + private sealed class RecordingTriggerRequestHandler(MessageRecorder recorder) + : IEventRequestHandler + { + public ValueTask HandleAsync(TriggerRequest request, CancellationToken cancellationToken) + { + recorder.Record(request); + return new(new TriggerResponse()); + } + } + // ========================================================================= // Test Infrastructure // ========================================================================= diff --git a/src/Mocha/test/Mocha.Sagas.Tests/SagaRouteConditionTests.cs b/src/Mocha/test/Mocha.Sagas.Tests/SagaRouteConditionTests.cs new file mode 100644 index 00000000000..63221cd0357 --- /dev/null +++ b/src/Mocha/test/Mocha.Sagas.Tests/SagaRouteConditionTests.cs @@ -0,0 +1,123 @@ +using Microsoft.Extensions.DependencyInjection; +using Mocha.Transport.InMemory; + +namespace Mocha.Sagas.Tests; + +/// +/// Tests for the route conditions that derives. Reply transitions are +/// gated on the saga-id header so non saga replies on the shared reply endpoint cannot select a saga; +/// typed replies additionally keep their message type term, while subscribe transitions route by type +/// alone. +/// +public class SagaRouteConditionTests +{ + [Fact] + public void Configure_Should_GateOnSagaIdOnly_When_OnAnyReply() + { + // arrange & act + var runtime = CreateRuntime(b => b.AddSaga()); + + // assert - OnAnyReply routes every saga-id reply to the consumer, so it gates on the saga-id alone + var route = GetSagaRoute(runtime, InboundRouteKind.Reply); + var description = route.Condition.Describe(); + Assert.Equal("HeaderPresent", description.Kind); + Assert.Equal("saga-id", description.Detail); + } + + [Fact] + public void Configure_Should_KeepMessageTypeTerm_When_TypedOnReply() + { + // arrange & act + var runtime = CreateRuntime(b => b.AddSaga()); + + // assert - a typed reply keeps its message type term in addition to the saga-id gate + var route = GetSagaRoute(runtime, InboundRouteKind.Reply); + var description = route.Condition.Describe(); + Assert.Equal("And", description.Kind); + Assert.Collection( + description.Children, + c => + { + Assert.Equal("HeaderPresent", c.Kind); + Assert.Equal("saga-id", c.Detail); + }, + c => Assert.Equal("MessageType", c.Kind)); + } + + [Fact] + public void Configure_Should_DeriveMessageTypeCondition_When_SubscribeTransition() + { + // arrange & act + var runtime = CreateRuntime(b => b.AddSaga()); + + // assert - the start event route is not saga-id gated, it routes by type alone + var route = GetSagaRoute(runtime, InboundRouteKind.Subscribe); + Assert.IsType(route.Condition); + } + + private static InboundRoute GetSagaRoute(MessagingRuntime runtime, InboundRouteKind kind) + { + var consumer = runtime.Consumers.OfType().Single(); + return runtime.Router.GetInboundByConsumer(consumer).Single(r => r.Kind == kind); + } + + private static MessagingRuntime CreateRuntime(Action configure) + { + var services = new ServiceCollection(); + services.AddInMemorySagas(); + var builder = services.AddMessageBus(); + configure(builder); + builder.AddInMemory(); + + var provider = services.BuildServiceProvider(); + return (MessagingRuntime)provider.GetRequiredService(); + } + + public sealed class ReplyState : SagaStateBase; + + public sealed class StartEvent; + + public sealed record Response; + + public sealed record Request : IEventRequest; + + /// + /// A saga that sends a request and finalizes on any reply (OnReply<object>). + /// + public sealed class AnyReplySaga : Saga + { + protected override void Configure(ISagaDescriptor descriptor) + { + descriptor + .Initially() + .OnEvent() + .StateFactory(_ => new ReplyState()) + .Send((_, _) => new Request()) + .TransitionTo("Awaiting"); + + descriptor.During("Awaiting").OnAnyReply().TransitionTo("Done"); + + descriptor.Finally("Done"); + } + } + + /// + /// A saga that sends a request and finalizes on a typed reply (OnReply<Response>). + /// + public sealed class TypedReplySaga : Saga + { + protected override void Configure(ISagaDescriptor descriptor) + { + descriptor + .Initially() + .OnEvent() + .StateFactory(_ => new ReplyState()) + .Send((_, _) => new Request()) + .TransitionTo("Awaiting"); + + descriptor.During("Awaiting").OnReply().TransitionTo("Done"); + + descriptor.Finally("Done"); + } + } +} diff --git a/src/Mocha/test/Mocha.Tests/MessageTypes/Conditions/RouteConditionTests.cs b/src/Mocha/test/Mocha.Tests/MessageTypes/Conditions/RouteConditionTests.cs new file mode 100644 index 00000000000..45f6181c518 --- /dev/null +++ b/src/Mocha/test/Mocha.Tests/MessageTypes/Conditions/RouteConditionTests.cs @@ -0,0 +1,281 @@ +using System.Collections.Immutable; +using System.Reflection; +using Mocha.Tests.Middlewares.Receive; + +namespace Mocha.Tests.MessageTypes.Conditions; + +/// +/// Tests for the route condition family that decides whether an inbound route selects its consumer +/// for a received message. +/// +public class RouteConditionTests : ReceiveMiddlewareTestBase +{ + private static readonly ContextDataKey s_correlationKey = new("test-correlation"); + private static readonly ContextDataKey s_sagaIdKey = new("test-saga-id"); + + [Fact] + public void MessageTypeCondition_Should_Match_When_ContextTypeIsExactType() + { + // arrange + var messageType = CreateMessageType(); + var condition = new MessageTypeCondition(messageType); + var context = new StubReceiveContext { MessageType = messageType }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.True(matches); + } + + [Fact] + public void MessageTypeCondition_Should_Match_When_ContextTypeEnclosesRouteType() + { + // arrange - route targets the base type, the context carries a derived type + var baseType = CreateMessageType(); + var derivedType = CreateMessageType(enclosed: [baseType]); + var condition = new MessageTypeCondition(baseType); + var context = new StubReceiveContext { MessageType = derivedType }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.True(matches); + } + + [Fact] + public void MessageTypeCondition_Should_NotMatch_When_ContextTypeIsUnrelated() + { + // arrange + var routeType = CreateMessageType(); + var otherType = CreateMessageType(); + var condition = new MessageTypeCondition(routeType); + var context = new StubReceiveContext { MessageType = otherType }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + [Fact] + public void MessageTypeCondition_Should_NotMatch_When_RouteTypeIsObjectAndContextIsConcrete() + { + // arrange - the structural reason OnAnyReply (OnReply) cannot route by type: + // object is excluded from every concrete message's enclosed types + var objectType = CreateMessageType(); + var concreteType = CreateMessageType(); + var condition = new MessageTypeCondition(objectType); + var context = new StubReceiveContext { MessageType = concreteType }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + [Fact] + public void MessageTypeCondition_Should_NotMatch_When_ContextTypeIsNull() + { + // arrange + var condition = new MessageTypeCondition(CreateMessageType()); + var context = new StubReceiveContext { MessageType = null }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + [Fact] + public void MessageTypeCondition_Should_Match_When_OptionalAndContextTypeIsNull() + { + // arrange - an optional reply route still selects when the message has no resolved type + var condition = new MessageTypeCondition(CreateMessageType(), optional: true); + var context = new StubReceiveContext { MessageType = null }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.True(matches); + } + + [Fact] + public void MessageTypeCondition_Should_NotMatch_When_StrictAndContextTypeIsNull() + { + // arrange - a strict route requires a resolved type to match + var condition = new MessageTypeCondition(CreateMessageType(), optional: false); + var context = new StubReceiveContext { MessageType = null }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + [Fact] + public void MessageTypeCondition_Should_NotMatch_When_OptionalAndWrongTypePresent() + { + // arrange - a present but unrelated type fails even for an optional route + var routeType = CreateMessageType(); + var otherType = CreateMessageType(); + var condition = new MessageTypeCondition(routeType, optional: true); + var context = new StubReceiveContext { MessageType = otherType }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + [Fact] + public void MessageTypeCondition_Should_Match_When_OptionalAndCorrectTypePresent() + { + // arrange + var routeType = CreateMessageType(); + var condition = new MessageTypeCondition(routeType, optional: true); + var context = new StubReceiveContext { MessageType = routeType }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.True(matches); + } + + [Fact] + public void MessageTypeCondition_Should_Match_When_OptionalAndEnclosedTypePresent() + { + // arrange - the optional route targets the base type, the context carries a derived type + var baseType = CreateMessageType(); + var derivedType = CreateMessageType(enclosed: [baseType]); + var condition = new MessageTypeCondition(baseType, optional: true); + var context = new StubReceiveContext { MessageType = derivedType }; + + // act + var matches = condition.Matches(context); + + // assert + Assert.True(matches); + } + + [Fact] + public void HeaderPresentCondition_Should_Match_When_HeaderPresent() + { + // arrange + var condition = new HeaderPresentCondition(s_correlationKey); + var context = new StubReceiveContext(); + context.Headers.Set(s_correlationKey, "abc"); + + // act + var matches = condition.Matches(context); + + // assert + Assert.True(matches); + } + + [Fact] + public void HeaderPresentCondition_Should_NotMatch_When_HeaderMissing() + { + // arrange + var condition = new HeaderPresentCondition(s_correlationKey); + var context = new StubReceiveContext(); + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + [Fact] + public void AndCondition_Should_Match_When_AllChildrenMatch() + { + // arrange + var condition = AndCondition.Create( + new HeaderPresentCondition(s_sagaIdKey), + new HeaderPresentCondition(s_correlationKey)); + var context = new StubReceiveContext(); + context.Headers.Set(s_sagaIdKey, "saga"); + context.Headers.Set(s_correlationKey, "abc"); + + // act + var matches = condition.Matches(context); + + // assert + Assert.True(matches); + } + + [Fact] + public void AndCondition_Should_NotMatch_When_FirstTermFails() + { + // arrange - the correlation header is present but the saga-id header is missing + var condition = AndCondition.Create( + new HeaderPresentCondition(s_sagaIdKey), + new HeaderPresentCondition(s_correlationKey)); + var context = new StubReceiveContext(); + context.Headers.Set(s_correlationKey, "abc"); + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + [Fact] + public void AndCondition_Should_NotMatch_When_SecondTermFails() + { + // arrange - the saga-id header is present but the correlation header is missing + var condition = AndCondition.Create( + new HeaderPresentCondition(s_sagaIdKey), + new HeaderPresentCondition(s_correlationKey)); + var context = new StubReceiveContext(); + context.Headers.Set(s_sagaIdKey, "saga"); + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + [Fact] + public void NoMatchCondition_Should_NeverMatch() + { + // arrange + var condition = NoMatchCondition.Instance; + var context = new StubReceiveContext { MessageType = CreateMessageType() }; + context.Headers.SetMessageKind(MessageKind.Reply); + + // act + var matches = condition.Matches(context); + + // assert + Assert.False(matches); + } + + private static MessageType CreateMessageType(ImmutableArray? enclosed = null) + { + var mt = new MessageType(); + SetPrivateProperty(mt, nameof(MessageType.Identity), "urn:message:Test"); + if (enclosed.HasValue) + { + SetPrivateProperty(mt, nameof(MessageType.EnclosedMessageTypes), enclosed.Value); + } + return mt; + } + + private static void SetPrivateProperty(object target, string propertyName, T value) + { + var property = target.GetType().GetProperty(propertyName, BindingFlags.Public | BindingFlags.Instance); + property!.SetValue(target, value); + } +} diff --git a/src/Mocha/test/Mocha.Tests/MessageTypes/InboundRouteTests.cs b/src/Mocha/test/Mocha.Tests/MessageTypes/InboundRouteTests.cs new file mode 100644 index 00000000000..a91e46386a1 --- /dev/null +++ b/src/Mocha/test/Mocha.Tests/MessageTypes/InboundRouteTests.cs @@ -0,0 +1,84 @@ +using Microsoft.Extensions.DependencyInjection; +using Mocha.Transport.InMemory; + +namespace Mocha.Tests.MessageTypes; + +/// +/// Tests for the condition that derives for a route, covering +/// the default type condition, the non-null invariant for all routes, and the no-match reply route. +/// +public class InboundRouteTests +{ + [Fact] + public void Initialize_Should_DeriveMessageTypeCondition_When_SubscribeRoute() + { + // arrange & act + var runtime = CreateRuntime(b => b.AddEventHandler()); + + // assert + var consumer = runtime.Consumers.First(c => c.Name == nameof(RouteEventHandler)); + var route = Assert.Single(runtime.Router.GetInboundByConsumer(consumer)); + Assert.IsType(route.Condition); + } + + [Fact] + public void Initialize_Should_DeriveMessageTypeCondition_When_RequestRoute() + { + // arrange & act + var runtime = CreateRuntime(b => b.AddRequestHandler()); + + // assert + var consumer = runtime.Consumers.First(c => c.Name == nameof(RouteRequestHandler)); + var route = Assert.Single(runtime.Router.GetInboundByConsumer(consumer)); + Assert.IsType(route.Condition); + } + + [Fact] + public void Initialize_Should_NeverLeaveConditionNull_When_RuntimeBuilt() + { + // arrange & act + var runtime = CreateRuntime(b => b.AddRequestHandler()); + + // assert + Assert.All(runtime.Router.InboundRoutes, r => Assert.NotNull(r.Condition)); + } + + [Fact] + public void Initialize_Should_DeriveNoMatchCondition_When_ReplyRouteHasNoMessageType() + { + // arrange & act + var runtime = CreateRuntime(b => b.AddRequestHandler()); + + // assert - the shared reply route carries no message type and never matches by the router + var replyRoute = Assert.Single(runtime.Router.InboundRoutes, r => r.Kind == InboundRouteKind.Reply); + Assert.Same(NoMatchCondition.Instance, replyRoute.Condition); + } + + private static MessagingRuntime CreateRuntime(Action configure) + { + var services = new ServiceCollection(); + var builder = services.AddMessageBus(); + configure(builder); + builder.AddInMemory(); + + var provider = services.BuildServiceProvider(); + return (MessagingRuntime)provider.GetRequiredService(); + } + + public sealed class RouteEvent; + + public sealed class RouteRequest : IEventRequest; + + public sealed class RouteResponse; + + public sealed class RouteEventHandler : IEventHandler + { + public ValueTask HandleAsync(RouteEvent message, CancellationToken cancellationToken) => default; + } + + public sealed class RouteRequestHandler : IEventRequestHandler + { + public ValueTask HandleAsync(RouteRequest request, CancellationToken cancellationToken) + => new(new RouteResponse()); + } +} diff --git a/src/Mocha/test/Mocha.Tests/Middlewares/Receive/RoutingMiddlewareTests.cs b/src/Mocha/test/Mocha.Tests/Middlewares/Receive/RoutingMiddlewareTests.cs index 9889dc75a3a..f5e447df7fa 100644 --- a/src/Mocha/test/Mocha.Tests/Middlewares/Receive/RoutingMiddlewareTests.cs +++ b/src/Mocha/test/Mocha.Tests/Middlewares/Receive/RoutingMiddlewareTests.cs @@ -6,11 +6,13 @@ namespace Mocha.Tests.Middlewares.Receive; /// -/// Tests for the RoutingMiddleware which routes incoming messages to the appropriate consumers -/// based on message type matching. +/// Tests for the RoutingMiddleware which selects consumers for the current endpoint by evaluating +/// each route's condition against the received message. /// public class RoutingMiddlewareTests : ReceiveMiddlewareTestBase { + private static readonly ContextDataKey s_correlationKey = new("test-correlation"); + [Fact] public async Task InvokeAsync_Should_AddConsumer_When_MessageTypeMatchesRoute() { @@ -105,7 +107,7 @@ public async Task InvokeAsync_Should_NotAddConsumers_When_MessageTypeIsNull() { // arrange var router = new MockMessageRouter(); - // Even if routes exist, null MessageType means no routing + // a type route does not match a message that never resolved a message type var consumer = new StubConsumer(); router.SetRoutes([CreateInboundRoute(CreateTestMessageType(), consumer)]); @@ -148,12 +150,12 @@ public async Task InvokeAsync_Should_MatchOnEnclosedMessageTypes() } [Fact] - public async Task InvokeAsync_Should_NotAddConsumer_When_RouteMessageTypeIsNull() + public async Task InvokeAsync_Should_NotAddConsumer_When_RouteHasNoMatchCondition() { - // arrange - route without a message type should not match + // arrange - a no-match route (the RPC reply route) never selects its consumer var messageType = CreateTestMessageType(); var consumer = new StubConsumer(); - var route = CreateInboundRoute(null, consumer); + var route = CreateInboundRoute(NoMatchCondition.Instance, consumer); var router = new MockMessageRouter(); router.SetRoutes([route]); @@ -170,6 +172,87 @@ public async Task InvokeAsync_Should_NotAddConsumer_When_RouteMessageTypeIsNull( Assert.Empty(feature.Consumers); } + [Fact] + public async Task InvokeAsync_Should_SelectHeaderRoute_When_MessageTypeIsNullButHeaderMatches() + { + // arrange - a header based route selects on envelope metadata alone, even with no message type + var consumer = new StubConsumer(); + var condition = new HeaderPresentCondition(s_correlationKey); + var route = CreateInboundRoute(condition, consumer); + + var router = new MockMessageRouter(); + router.SetRoutes([route]); + + var middleware = new RoutingMiddleware(router); + var context = new StubReceiveContext { MessageType = null }; + context.Headers.SetMessageKind(MessageKind.Reply); + context.Headers.Set(s_correlationKey, "abc"); + var next = CreatePassthroughDelegate(); + + // act + await middleware.InvokeAsync(context, next); + + // assert + var feature = context.Features.GetOrSet(); + Assert.Contains(consumer, feature.Consumers); + } + + [Fact] + public async Task InvokeAsync_Should_AddBothRoutes_When_TypeRouteAndHeaderRouteCoexist() + { + // arrange - a reply that matches both an ordinary type route and a header based reply route + var messageType = CreateTestMessageType(); + var typeConsumer = new StubConsumer(); + var sagaConsumer = new StubConsumer(); + var typeRoute = CreateInboundRoute(messageType, typeConsumer); + var sagaRoute = CreateInboundRoute(new HeaderPresentCondition(s_correlationKey), sagaConsumer); + + var router = new MockMessageRouter(); + router.SetRoutes([typeRoute, sagaRoute]); + + var middleware = new RoutingMiddleware(router); + var context = new StubReceiveContext { MessageType = messageType }; + context.Headers.SetMessageKind(MessageKind.Reply); + context.Headers.Set(s_correlationKey, "abc"); + var next = CreatePassthroughDelegate(); + + // act + await middleware.InvokeAsync(context, next); + + // assert + var feature = context.Features.GetOrSet(); + Assert.Equal(2, feature.Consumers.Count); + Assert.Contains(typeConsumer, feature.Consumers); + Assert.Contains(sagaConsumer, feature.Consumers); + } + + [Fact] + public async Task InvokeAsync_Should_AddConsumerOnce_When_TwoRoutesShareConsumer() + { + // arrange - the same consumer is bound to two matching routes; the consumer set dedups it + var messageType = CreateTestMessageType(); + var consumer = new StubConsumer(); + var route1 = CreateInboundRoute(messageType, consumer); + var route2 = CreateInboundRoute(new HeaderPresentCondition(s_correlationKey), consumer); + + var router = new MockMessageRouter(); + router.SetRoutes([route1, route2]); + + var middleware = new RoutingMiddleware(router); + var context = new StubReceiveContext { MessageType = messageType }; + context.Headers.SetMessageKind(MessageKind.Reply); + context.Headers.Set(s_correlationKey, "abc"); + var next = CreatePassthroughDelegate(); + + // act + await middleware.InvokeAsync(context, next); + + // assert + var feature = context.Features.GetOrSet(); + Assert.Single(feature.Consumers); + Assert.Contains(consumer, feature.Consumers); + } + private static MessageType CreateTestMessageType(ImmutableArray? enclosedTypes = null) { var mt = new MessageType(); @@ -180,11 +263,21 @@ private static MessageType CreateTestMessageType(ImmutableArray? en return mt; } - private static InboundRoute CreateInboundRoute(MessageType? messageType, Consumer? consumer) + private static InboundRoute CreateInboundRoute(MessageType messageType, Consumer consumer) + => CreateInboundRoute(new MessageTypeCondition(messageType), consumer, messageType); + + private static InboundRoute CreateInboundRoute(RouteCondition condition, Consumer consumer) + => CreateInboundRoute(condition, consumer, messageType: null); + + private static InboundRoute CreateInboundRoute( + RouteCondition condition, + Consumer consumer, + MessageType? messageType) { var route = new InboundRoute(); SetPrivateProperty(route, nameof(InboundRoute.MessageType), messageType); SetPrivateProperty(route, nameof(InboundRoute.Consumer), consumer); + SetPrivateProperty(route, nameof(InboundRoute.Condition), condition); return route; }