Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -294,7 +294,7 @@ private void ConfigureQueueEndpoint(RabbitMQQueueDescriptorConfiguration configu
target.ConsumerIdentities.AddRange(configuration.ConsumerIdentities);
target.ReceivedMessageTypes.AddRange(configuration.ReceivedMessageTypes);

target.BindMode = configuration.BindMode ?? target.BindMode ?? MessagingBindMode.Implicit;
target.BindMode = configuration.BindMode ?? target.BindMode;

if (configuration.Kind is not null && target.Kind == ReceiveEndpointKind.Default)
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,7 @@ private RabbitMQReceiveEndpointDescriptor(IMessagingConfigurationContext discove
Configuration = new RabbitMQReceiveEndpointConfiguration
{
Name = name,
QueueName = name,
BindMode = MessagingBindMode.Implicit
QueueName = name
};
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -270,13 +270,6 @@ public override void DiscoverTopology(

if (messageType.HasPerMessageRoutingKey())
{
if (autoBind)
{
throw ThrowHelper.CannotCreateRabbitMQAutoBind(
messageType,
"it has a per-message routing key");
}

continue;
}
Comment thread
PascalSenn marked this conversation as resolved.

Expand Down Expand Up @@ -306,6 +299,16 @@ public override void DiscoverTopology(
// exchange into the receive queue. This keeps Publish<T> and Send<T> converged on the
// same queue while still allowing separate publish and send exchange names.
// Example: publish/order-created -> send/order-created -> queue/orders.
//
// Skip the whole convention pair when the endpoint binds explicitly: the upstream
// publish/send exchanges only exist to feed the exchange-to-queue binding below, so
// without that final hop they would be declared on this service but bound to nothing,
// contradicting the "explicit means explicit" contract.
if (!autoBind)
{
continue;
}

var publishExchangeName = context.Naming.GetPublishEndpointName(messageType.RuntimeType);
_topology.EnsureExchange(publishExchangeName);

Expand All @@ -321,10 +324,7 @@ public override void DiscoverTopology(
static (_, _, _) => new RabbitMQBindingConfiguration());
}

if (autoBind)
{
_topology.EnsureExchangeToQueueBinding(sendExchangeName, rabbitConfiguration.QueueName);
}
_topology.EnsureExchangeToQueueBinding(sendExchangeName, rabbitConfiguration.QueueName);
}
}

Expand All @@ -340,14 +340,21 @@ public override void DiscoverTopology(
return;
}

if (rabbitConfiguration.ExchangeName is not null)
// Under BindExplicitly, the dispatch endpoint must not materialize a destination
// exchange the user never declared: the user retains full ownership of the topology,
// and a default-config (fanout) shadow can collide with a previously-declared exchange
// on the broker (PRECONDITION_FAILED on declare). The user is expected to either
// DeclareExchange it explicitly or rely on a pre-existing broker entity.
var bindImplicitly = Transport.BindMode == MessagingBindMode.Implicit;

if (rabbitConfiguration.ExchangeName is not null && bindImplicitly)
{
_topology.GetOrAddExchange(
rabbitConfiguration.ExchangeName,
static _ => new RabbitMQExchangeConfiguration());
}

if (rabbitConfiguration.QueueName is not null)
if (rabbitConfiguration.QueueName is not null && bindImplicitly)
{
_topology.GetOrAddQueue(
rabbitConfiguration.QueueName,
Expand All @@ -357,7 +364,7 @@ public override void DiscoverTopology(
var schema = Transport.Schema;

if (rabbitConfiguration.ExchangeName is not null
&& Transport.BindMode == MessagingBindMode.Implicit)
&& bindImplicitly)
{
foreach (var (runtimeType, kind) in rabbitConfiguration.Routes)
{
Expand Down
184 changes: 184 additions & 0 deletions src/Mocha/src/Mocha/Builder/MessageBusBuilder.Validation.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
using System.Text;

namespace Mocha;

public partial class MessageBusBuilder
{
private static void ValidateInboundRoutesAreBound(IMessagingSetupContext context)
{
var unboundRoutes = context.Router.InboundRoutes
.Where(route =>
route is { IsInitialized: true, IsCompleted: false, Endpoint: null, Kind: not InboundRouteKind.Reply })
.OrderBy(route => route.Consumer?.Name, StringComparer.Ordinal)
.ThenBy(route => route.MessageType?.RuntimeType.FullName, StringComparer.Ordinal)
.ThenBy(route => route.Kind)
.ToArray();

if (unboundRoutes.Length == 0)
{
return;
}

throw new InvalidOperationException(context.FormatUnboundInboundRoutes(unboundRoutes));
}
}

file static class Extensions
{
public static string FormatUnboundInboundRoutes(
this IMessagingSetupContext context,
IReadOnlyList<InboundRoute> routes)
{
var sb = new StringBuilder();
sb.Append("The message bus configuration has ");
sb.Append(routes.Count);
sb.Append(routes.Count == 1 ? " unbound inbound route." : " unbound inbound routes.");
sb.AppendLine();
sb.AppendLine("Each route below was registered but never attached to a receive endpoint.");

for (var i = 0; i < routes.Count; i++)
{
var route = routes[i];
sb.AppendLine();
sb.Append('[');
sb.Append(i + 1);
sb.Append("] ");
sb.Append(route.GetConsumerName());
sb.Append(" (");
sb.Append(route.GetConsumerType());
sb.AppendLine(")");

sb.Append(" message : ");
sb.AppendLine(route.GetMessage());
sb.Append(" kind : ");
sb.AppendLine(route.Kind.ToString());
sb.Append(" where : ");
sb.AppendLine(route.GetCondition());
sb.Append(" fix : ");
sb.AppendLine(route.GetHint(context));
}

return sb.ToString();
}

private static string GetHint(this InboundRoute route, IMessagingSetupContext context)
{
var sameMessageRoute = route.FindBoundRouteForSameMessage(context);
if (sameMessageRoute is not null)
{
return "Another route for this message is already bound to consumer "
+ $"'{sameMessageRoute.GetConsumerName()}' on receive endpoint "
+ $"'{sameMessageRoute.Endpoint!.Name}' for transport "
+ $"'{sameMessageRoute.Endpoint.Transport.Name}'. "
+ "If this consumer is unintended, remove its registration. "
+ "If it should receive independently, bind it to its own receive endpoint.";
}

var transportHint = GetTransportHint(context);
var endpointHint = GetEndpointHint(context);

return transportHint
+ " If this consumer should receive over a transport, bind it to a receive endpoint. "
+ "If it is handled in-process only or superseded, remove it from message bus registration. "
+ endpointHint;
}

private static InboundRoute? FindBoundRouteForSameMessage(this InboundRoute route, IMessagingSetupContext context)
{
if (route.MessageType is not { } messageType)
{
return null;
}

return context.Router
.GetInboundByMessageType(messageType)
.Where(candidate =>
!ReferenceEquals(candidate, route)
&& candidate.Kind == route.Kind
&& candidate.Endpoint is not null)
.OrderBy(candidate => candidate.Consumer?.Name, StringComparer.Ordinal)
.ThenBy(candidate => candidate.Endpoint?.Name, StringComparer.Ordinal)
.FirstOrDefault();
}

private static string GetTransportHint(IMessagingSetupContext context)
{
if (context.Transports.Length == 0)
{
return "No messaging transports are configured, so no receive endpoint can be discovered.";
}

if (context.Transports.Length == 1)
{
var transport = context.Transports[0];
return transport.BindMode == MessagingBindMode.Explicit
? $"Transport '{transport.Name}' uses explicit bind mode, so no convention receive endpoint is created for unbound consumers."
: $"Transport '{transport.Name}' uses implicit bind mode, but it did not connect this route.";
}

var transports = string.Join(
", ",
context.Transports.Select(static t => $"'{t.Name}' ({t.BindMode})"));

return $"No configured transport connected this route. Configured transports: {transports}.";
}

private static string GetEndpointHint(IMessagingSetupContext context)
{
var endpoints = context.Transports
.Select(static transport => new
{
Transport = transport.Name,
Endpoints = transport.ReceiveEndpoints
.Where(static endpoint => endpoint.Kind == ReceiveEndpointKind.Default)
.Select(static endpoint => endpoint.Name)
.Order(StringComparer.Ordinal)
.ToArray()
})
.Where(static transport => transport.Endpoints.Length > 0)
.OrderBy(static transport => transport.Transport, StringComparer.Ordinal)
.ToArray();

if (endpoints.Length == 0)
{
return "Available default receive endpoints: none.";
}

return "Available default receive endpoints: "
+ string.Join("; ", endpoints.Select(static t => $"{t.Transport}: {string.Join(", ", t.Endpoints)}"))
+ ".";
}

private static string GetConsumerName(this InboundRoute route)
=> route.Consumer?.Name ?? "(unknown consumer)";

private static string GetConsumerType(this InboundRoute route)
=> route.Consumer?.Identity.FullName
?? route.Consumer?.Identity.Name
?? route.Consumer?.GetType().FullName
?? "(unknown consumer type)";

private static string GetMessage(this InboundRoute route)
{
if (route.MessageType is null)
{
return "(no message type)";
}

return route.MessageType.Identity + " (" + route.MessageType.RuntimeType.FullName + ")";
}

private static string GetCondition(this InboundRoute route)
=> FormatCondition(route.Condition.Describe());

private static string FormatCondition(RouteConditionDescription condition)
{
var text = condition.Detail is null
? condition.Kind
: $"{condition.Kind}({condition.Detail})";

return condition.Children.Count == 0
? text
: text + "[" + string.Join(", ", condition.Children.Select(FormatCondition)) + "]";
}
}
2 changes: 2 additions & 0 deletions src/Mocha/src/Mocha/Builder/MessageBusBuilder.cs
Original file line number Diff line number Diff line change
Expand Up @@ -522,6 +522,8 @@ public MessagingRuntime Build(IServiceProvider applicationServices)
}
}

ValidateInboundRoutesAreBound(setupContext);

foreach (var route in router.InboundRoutes)
{
if (!route.IsCompleted)
Expand Down
2 changes: 1 addition & 1 deletion src/Mocha/src/Mocha/MessageTypes/InboundRoute.cs
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,7 @@ public void Complete(IMessagingConfigurationContext context)

if (Endpoint is null)
{
throw ThrowHelper.RouteEndpointNotConnected();
throw ThrowHelper.RouteEndpointNotConnected(this);
}

MarkCompleted();
Expand Down
2 changes: 1 addition & 1 deletion src/Mocha/src/Mocha/MessageTypes/OutboundRoute.cs
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ public void Complete(IMessagingConfigurationContext context)

if (Endpoint is null)
{
throw ThrowHelper.RouteEndpointNotConnected();
throw ThrowHelper.RouteEndpointNotConnected(this);
}

Destination ??= Endpoint.Address;
Expand Down
36 changes: 25 additions & 11 deletions src/Mocha/src/Mocha/ThrowHelper.cs
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,31 @@ public static Exception BeforeAndAfterConflict()
public static Exception MiddlewareKeyNotFound(string key)
=> new InvalidOperationException($"Middleware with key {key} not found");

public static Exception RouteEndpointNotConnected()
=> new InvalidOperationException("Endpoint is not connected");
public static Exception RouteEndpointNotConnected(InboundRoute route)
{
var consumer = route.Consumer?.Identity.FullName
?? route.Consumer?.Name
?? "(unknown consumer)";
var message = route.MessageType?.RuntimeType.FullName
?? route.MessageType?.Identity
?? "(no message type)";

return new InvalidOperationException(
$"Inbound route for consumer '{consumer}', message '{message}', and kind '{route.Kind}' "
+ "is not connected to a receive endpoint.");
}

public static Exception RouteEndpointNotConnected(OutboundRoute route)
{
var message = route.MessageType?.RuntimeType.FullName
?? route.MessageType?.Identity
?? "(no message type)";
var destination = route.Destination?.ToString() ?? "(no destination)";

return new InvalidOperationException(
$"Outbound route for message '{message}', kind '{route.Kind}', and destination '{destination}' "
+ "is not connected to a dispatch endpoint.");
}

public static Exception RouteMustNotBeInitialized()
=> new InvalidOperationException("Route must not be initialized");
Expand Down Expand Up @@ -87,15 +110,6 @@ public static Exception NoTransportForMessageType(MessageType messageType)
=> new InvalidOperationException(
$"No transport can handle message type '{messageType.RuntimeType.FullName}'.");

public static InvalidOperationException CannotCreateRabbitMQAutoBind(
MessageType messageType,
string reason)
=> new(
"Cannot create a RabbitMQ auto-bind for consumed message type "
+ $"'{messageType.RuntimeType.FullName}' because "
+ reason
+ ".");

public static Exception MultipleDefaultTransports(IEnumerable<string> transportNames)
=> new InvalidOperationException(
$"Multiple transports are flagged as default: {string.Join(", ", transportNames.Select(t => $"'{t}'"))}. "
Expand Down
8 changes: 8 additions & 0 deletions src/Mocha/src/Mocha/Transport/RoutingStrategy.cs
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,14 @@ protected virtual void DiscoverReplyEndpoints(IMessagingSetupContext context)

Transport.ReplyDispatchEndpoint = Transport.DispatchEndpoints.FirstOrDefault(x => x.Kind == DispatchEndpointKind.Reply);
Transport.ReplyReceiveEndpoint = Transport.ReceiveEndpoints.FirstOrDefault(x => x.Kind == ReceiveEndpointKind.Reply);

if (Transport.ReplyReceiveEndpoint is { } replyEndpoint)
{
foreach (var route in context.Router.InboundRoutes.Where(static route => route.Kind == Reply).ToArray())
{
context.BindRouteToEndpoint(route, replyEndpoint);
}
}
}

protected virtual void DiscoverImplicitEndpoints(IMessagingSetupContext context)
Expand Down
Loading
Loading