How to correctly configure publishing messages to a topic exchange using CommunityToolkit.Aspire.MassTransit.RabbitMQ?

[denis-bredun-work]

I have two separate projects – a publisher and a consumer. I’m using MassTransit, and to simplify configuration I’m using the NuGet package CommunityToolkit.Aspire.MassTransit.RabbitMQ.

My publisher setup:

builder.AddMassTransitRabbitMq(
    "rabbitmq",
    options => { options.DisableTelemetry = false; },
    consumers => { }
);

My consumer setup:

builder.AddMassTransitRabbitMq(
    "rabbitmq",
    options => { options.DisableTelemetry = false; },
    consumers =>
    {
        consumers.AddConsumer<MyConsumer, MyConsumerDefinition>();
    }
);

Consumer:

public class MyConsumer() : IConsumer<Message>
{
    public async Task Consume(ConsumeContext<Message> context)
    {
        // ...
    }
}

Consumer definition:

public class MyConsumerDefinition : ConsumerDefinition<MyConsumer>
{
    private readonly ExchangeOptions _exchange;

    public MyConsumerDefinition(IOptions<Dictionary<string, ExchangeOptions>> exchangeConfig)
    {
        _exchange = exchangeConfig.Value["SomeExchange"];
        EndpointName = _exchange.QueueName;
    }

    protected override void ConfigureConsumer(
        IReceiveEndpointConfigurator endpointConfigurator,
        IConsumerConfigurator<MyConsumer> consumerConfigurator,
        IRegistrationContext context)
    {
        endpointConfigurator.UseMessageRetry(r =>
        {
            r.Interval(_exchange.RetryCount, TimeSpan.FromSeconds(_exchange.RetryIntervalSeconds));
        });

        if (endpointConfigurator is IRabbitMqReceiveEndpointConfigurator rabbit)
        {
            rabbit.Durable = _exchange.Durable;
            rabbit.AutoDelete = _exchange.AutoDelete;

            rabbit.Bind(_exchange.Name, x =>
            {
                x.ExchangeType = _exchange.Type;
                x.RoutingKey = _exchange.RoutingKey;
            });
        }
    }
}

Publishing messages:

await publisher.PublishBatch(messages, ctx =>
{
    ctx.DestinationAddress = new Uri($"exchange:{_exchange.Name}?type={_exchange.Type}");
    ctx.SetRoutingKey($"{ctx.Message.Part1.ToLower()}.{ctx.Message.Part2.ToLower()}");
}, ct);

When I check RabbitMQ management, I see the following setup:

Path:Message        fanout
nameOfExchange      topic
nameOfQueue.queue   fanout

This indicates that MassTransit automatically creates an exchange and a queue for the message type (PostPlatformMessage) as fanout, even though I configured a topic exchange. As a result, the consumer receives all messages of this type regardless of the routing key.

I want messages to be routed based on the RoutingKey to the correct consumer, so that only messages matching the consumer’s configured routing key are delivered to it, using a topic exchange.

Currently, my consumer receives messages that should be routed to other consumers based on the routing key. Even though I configure the exchange as topic and set routing keys for both publishing and binding, messages seem to be delivered to the consumer regardless of the routing key.

Could someone advise why this might be happening and how to correctly configure MassTransit + Aspire + RabbitMQ to respect routing keys when publishing to a topic exchange?