How do you configure IRequestClient<T> and IConsumer<T> using MassTransit.Multibus?

Viewed 1140

We have been using MassTransit with a single RabbitMq transport that is internal to our services. We have a new RabbitMq server that is public that we also want to connect to for certain events, so naturally we want to use the Multibus feature. The connections are successful, and messages seem to publish fine, but our old RequestClient consumers no longer appear to be working on the original bus, and I am not sure why. The error thrown says MassTransit.RequestTimeoutException: Timeout waiting for response. Multibus IBuses should start on their own, correct?

Here is what it looks like in Startup.cs ConfigureServices (ICorrespondenceInternalBus and ICorrespondenceExternalBus both inherit from IBus):

...
//First bus
services.AddMassTransit<ICorrespondenceInternalBus>(c =>
{
    c.AddConsumersFromNamespaceContaining(GetType());
    ConfigureAdditionalMassTransitServices(c);

    c.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(new Uri($"rabbitmq://{rabbitMqServerName}:/"),
            h =>
            {
                h.Username("guest");
                h.Password("guest");
            });

        cfg.ConfigureEndpoints(context, new CorrespondenceSystemEndpointNameFormatter());
        cfg.UseMessageRetry(retryConfig => retryConfig.Interval(5, TimeSpan.FromMilliseconds(250)));
        cfg.UseHealthCheck(context);
    });
});
services.AddMassTransitHostedService();
...
//second bus
services.AddMassTransit<ICorrespondenceExternalBus>(c =>
{
    c.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(rabbitMqServerName, port, virtualHost, h =>
        {
            h.Username(username);
            h.Password(password);
            if (useSsl)
                h.UseSsl(s => s.Protocol = SslProtocols.Tls12);
        });
        cfg.MessageTopology.SetEntityNameFormatter(new CorrespondenceSystemExternalEntityNameFormatter());
        cfg.UseHealthCheck(context);
    });
});

In the above, both of the buses register and the Exchanges in rabbitmq appear to receive published messages. The part that is not working is consuming messages from RequestClients. Here is how the RequestClients are being registered:

protected override void ConfigureAdditionalMassTransitServices(
IServiceCollectionConfigurator<ICorrespondenceInternalBus> configurator)
{
    configurator.AddRequestClient<ICheckForDuplicateQuery>();
}

The RequestHandler in action:

public class Handler : IRequestHandler<Command, Dto>
{
    private readonly IRequestClient<ICheckForDuplicateQuery> _duplicateCheckClient;

    public Handler(IRequestClient<ICheckForDuplicateQuery> duplicateCheckClient)
    {
        _duplicateCheckClient = duplicateCheckClient;
    }

    public async Task<Dto> Handle(Command request, CancellationToken cancellationToken)
    {
        var duplicateQuery = new Query();
        var duplicateCheckResult = await _duplicateCheckClient.GetResponse<ICheckForDuplicateQueryResult>(duplicateQuery, cancellationToken, TimeSpan.FromSeconds(10));

        if (duplicateCheckResult.Message.IsDuplicate)
            return new DuplicateDto(duplicateCheckResult.Message.CorrelationIds.First());

        ...
    }
}

And finally the consumer:

public class CheckForDuplicateQueryHandler : IConsumer<ICheckForDuplicateQuery>
{    
    ...
    public async Task Consume(ConsumeContext<ICheckForDuplicateQuery> context)
    {
        if (context is null)
            throw new ArgumentNullException(nameof(context));
        ...
        await context.RespondAsync(new Result()).ConfigureAwait(false);
    }

    private class Result : ICheckForDuplicateQueryResult
    {
        ...
    }
}

The consumer never enters and the request client times out. For comparison, here is what everything looked like before we attempted Multibus when the RequestClients worked fine (the consumer and request client logic are exactly the same, only the Startup.cs is different: Previous (single bus) Startup.cs:

services.AddMassTransit(c =>
{
    c.AddConsumersFromNamespaceContaining(GetType());
    ConfigureAdditionalMassTransitServices(c);

    c.AddBus(provider => Bus.Factory.CreateUsingRabbitMq(sbc =>
    {
        sbc.Host(new Uri($"rabbitmq://{rabbitMqServerName}:/"),
            h =>
            {
                h.Username("guest");
                h.Password("guest");
            });

        sbc.ConfigureEndpoints(provider, new CorrespondenceSystemEndpointNameFormatter());
        sbc.UseMessageRetry(cfg => cfg.Interval(5, TimeSpan.FromMilliseconds(250)));
    }));
});
HealthChecksBuilder.AddRabbitMqHealthcheck(rabbitMqServerName);
...
public virtual void Configure(IApplicationBuilder app, IHostEnvironment env, IHostApplicationLifetime appLifetime, IBusControl bus)
{
    ...
    appLifetime.ApplicationStarted.Register(bus.Start);
    appLifetime.ApplicationStopping.Register(bus.Stop);
}
...
//registering the RequestClients previously:
protected override void ConfigureAdditionalMassTransitServices(IServiceCollectionBusConfigurator configurator)
{
    configurator.AddRequestClient<ICheckForDuplicateQuery>();
}

Thanks in advance for any help! If you need to see more code snippets I'm glad to provide them, I was trying to keep it concise with only what is needed/affected in the changes.

1 Answers

I have confirmed that the request client should be using the correct bus instance, depending upon where it was configured in this unit test commit.

So, I'm not sure why you aren't seeing the same behavior.

Related