MultiBus setup - Configuring a singular bus to set the entity name formatter

[elajeunesse]

We are currently using a MultiBus solution for MassTransit and we are running into issues where the configuration of one Bus is being used by another bus.

services.AddMassTransit(config =>
{
    config.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(
            configuration.HostName,
            configuration.VHostName,
            cfgHost =>
            {
                cfgHost.Username(configuration.Username);
                cfgHost.Password(configuration.Password);
            }
        );

        cfg.ConfigureEndpoints(context);
    });
});

services.AddMassTransit<SecondBus>(config =>
{
    config.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(
            configuration.HostName,
            configuration.VHostNameTwo,
            cfgHost =>
            {
                cfgHost.Username(configuration.Username);
                cfgHost.Password(configuration.Password);
            }
        );

        cfg.MessageTopology.SetEntityNameFormatter(new Formatter());

        cfg.ConfigureEndpoints(context);
    });

public class Formatter: IEntityNameFormatter
{
    public string FormatEntityName<T>()
    {
        return typeof(T).FullName ?? "";
    }
}

When I publish a message using the default bus, it's using the Formatter when we don't want it to. Is there any reason for this, or any alternative ways we can do this?

Similar issue as described here: https://stackoverflow.com/questions/77960200/masstransit-multibus-entity-name-formaters

[phatboyg]

The message topology is shared across all bus instances, so there isn't a workaround at this point. When MultiBus was added, I don't think this was considered, and I'm not sure what the implications of splitting it off would be at this point.

[ardent42c]

thanks for the reply. This forces separate application instance for each install, which is not very efficient and harder to maintain.
Any chance of something like that implemented in the near future?
Another possible solution is maybe to allow overriding the topic name directly Under UsingAmazonSqs - something like:
sqsBusFactoryConfigurator.Topic<MyMessage>(c => c.Name = "Topic-For-Server1";)

[phatboyg]

Easy enough.
be8e177

[ardent42c]

Wow. Thanks!
Will try it out with a local build.

[phatboyg]

The pre-release packages should already be on NuGet.

[ardent42c]

Even better, thanks.

[ardent42c]

I've finally managed to test this. Have updated all MassTransit nugets to to version 8.2.3-develop.1818 and retried.
My IEntityNameFormatter class was indeed called for each bus for each of the IConsumer, and I have verified that it did return a completely different result in the FormatEntityName function for each IConsumer in each different Bus (used a prefix for each server plus the typeof(T) of the consumer type). This IS different than the previous version, where the FormatEntityName was only called once for each consumer, and not also for each bus.
HOWEVER, when I then checked AWS SQS, I found the the queues were subscribed to the wrong topic.
The following image shows the problem (it has two subscription, since I previously ran each server separately, which registered to the correct topic). The different servers are QA and SOURCE. Source was added with AddMassTransit and QA added with AddMassTransit<IBUS1>

[ardent42c]

So there's nothing I could do? Any way to separate the duplicate consumers maybe with generics or some other way?

[phatboyg]

You could set the endpoint name formatter with a prefix instead of using .Endpoint(name)

cfg.ConfigureEndpoint(context, new KebabCaseEndpointNameFormatter("server1", false));

Because the registration isn't bus specific either.

[ardent42c]

I don't seem to find that function (ConfigureEndpoint) anywhere. Can you point out what exactly is cfg here?
In the sample there is this part
busRegistrationConfigurator.SetEndpointNameFormatter(new KebabCaseEndpointNameFormatter(prefix: server2Name, includeNamespace: false)); (which was redundant as it was overwritten by the .Endpoint(name) bit)
but busRegistrationConfigurator, nor cfg in AddConsumer seem to have ConfigureEndpoint.
The sqsBusFactoryConfigurator has ConfigureEndpoints(context) but that's not the same.

[ardent42c]

I guess you meant sqsBusFactoryConfigurator.ConfigureEndpoints ?
I tried to move the KebabCaseEndpointNameFormatter with different prefixes for each server to the ConfigureEndpoints(context, new KebabCaseEndPointNameFormatter..), and also tried to remove busRegistrationConfigurator.SetEndpointNameFormatter and of course the .Endpoint, nothing helped. Still only creates the queue for the first server.

[phatboyg]

This works for me:

[Test]
public async Task Should_work()
{
    await using var provider = new ServiceCollection()
        .AddMassTransitTestHarness(x =>
        {
            x.AddConsumer<MyMessageConsumer>(cfg => cfg.ConcurrentMessageLimit = 1);

            x.UsingAmazonSqs((context, cfg) =>
            {
                cfg.LocalstackHost();

                cfg.MessageTopology.SetEntityNameFormatter(new BusEnvironmentNameFormatter("server1"));

                cfg.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("server1", false));
            });
        })
        .AddMassTransit<ISecondBus>(x =>
        {
            x.AddConsumer<MyMessageConsumer>(cfg => cfg.ConcurrentMessageLimit = 1);

            x.UsingAmazonSqs((context, cfg) =>
            {
                cfg.LocalstackHost();

                cfg.MessageTopology.SetEntityNameFormatter(new BusEnvironmentNameFormatter("server2"));

                cfg.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("server2", false));
            });
        })
        .BuildServiceProvider(true);

    var harness = await provider.StartTestHarness();

    var bus = harness.Bus;

    Assert.That(bus.Topology.Publish<MyMessage>().TryGetPublishAddress(bus.Address, out var publishAddress), Is.True);
    Assert.That(publishAddress, Is.EqualTo(new Uri(bus.Address, "server1-topic-MyMessage?type=topic")));

    bus = provider.GetRequiredService<ISecondBus>();

    Assert.That(bus.Topology.Publish<MyMessage>().TryGetPublishAddress(bus.Address, out publishAddress), Is.True);
    Assert.That(publishAddress, Is.EqualTo(new Uri(bus.Address, "server2-topic-MyMessage?type=topic")));
}

[phatboyg]

Added this commit to separate the endpoint name formatter registration(s):
f1cfc7a

[ardent42c]

I'm not sure what exactly I'm doing wrong. I'm building the service provider, then getting the IBusControl and starting it.
AFAIK, there is no second IBusControl. How am I supposed to separately start the second bus?
This is my code (after adding the primary and secondary mass transit to the serviceCollection):
var sp = serviceCollection.BuildServiceProvider();
var GlobalBusController = sp.GetService<IBusControl>()!;
var bus1 = sp.GetService<IBus>();
var bus2 = sp.GetRequiredService<ISecondBus>(); //var bus2 = sp.GetService<ISecondBus>(); (replaced to GetRequiredService to match your code)
await GlobalBusController.StartAsync();

[phatboyg]

That isn't how it works.

var hostedServices = sp.GetServices<IHostedService>();
foreach(var service in hostedServices)
await service.StartAsync();

[ardent42c]

You are of course correct. I'm a bit new to services, and I did copy the start from some base example.
Only had to .AddLogging() but other than that it worked great! even in .NET 4.7
I can see the two queues created, and each is subscribed to the correct topic!
Thank you very much for your help and patience.

The only work-around that I still need to use for using the same bus for multiple servers is the ability to add buses from configuration data and not fixed in code (aka have a list of n servers, each has a name, region, accesskey and secret key and each should register the same consumers like the above example which is fixed for 2 servers).
My current workaround is to create a fixed list of interfaces IBus1 to IBusX, so n is known to be less then X, and use something like:

                switch (configNum)
                {
                    case 0:
                        serviceCollection.AddMassTransit(ConfigMT)
                        break;
                    case 1:
                        serviceCollection.AddMassTransit<IBus1>(ConfigMT);
                        break;
                    case 2:
                        serviceCollection.AddMassTransit<IBus2>(ConfigMT);
                        break;
                    ...
                }

which is ugly as hell. I did find a reflection based solution (to create types at runtime, and call AddMassTransit with reflection), but that's ugly as well.
Would it break the mold too much to be able to add additional buses by names instead of using generics with inherited types?

[phatboyg]

There is no way to use a string or anything other than an interface to configure additional bus instances.

[ardent42c]

Not a problem.
I've updated my sample to use a simple extension I wrote (without reflection) to support just that (as I said, it's a bit ugly, but does it's job)
MassTransitMultiBusCoreTest