CODEWITHSUNDEEP
c#asp.net-core.net-coremicroservicesmasstransit
Bulchsu
Bulchsu

Reputation: 700

How to share message definition across microservices using MassTransit with RabbitMQ?

I am new to microservices and i have a little problem with connecting the publisher with the subscriber using the MassTransit framework. I have an example integration event generated in one of the services when the user is created. Here's the definition:

public sealed record UserCreatedIntegrationEvent : IntegrationEvent
{
    public UserCreatedIntegrationEvent(Guid id,
        string login,
        string firstName,
        string lastName,
        string mailAddress)
            : base(id,
                nameof(UserCreatedIntegrationEvent))
    {
        Login = login;
        FirstName = firstName;
        LastName = lastName;
        MailAddress = mailAddress;
    }

    public string Login { get; }
    public string FirstName { get; }
    public string LastName { get; }
    public string MailAddress { get; }
}

The message is contained in the User.Application project. The message is properly being published on the message broker bus. Now i need to receive an event in the other service, so i need to know the definition of UserCreatedIntegrationEvent there, to deserialize it etc. I could refer the User.Application project but in my opinion it could cause some problems and in general this approach violates microservice autonomy rules i think. The other solution is to duplicate the message definition in the receiver service. Then i copy the UserCreatedIntegrationEvent and connect new pasted event to the specific handler:

public abstract class IntegrationEventHandler<TIntegrationEvent> : IConsumer<TIntegrationEvent>
    where TIntegrationEvent : IntegrationEvent
{
    protected ConsumeContext<TIntegrationEvent> ConsumeContext { get; private set; }

    public async Task Consume(ConsumeContext<TIntegrationEvent> context)
    {
        ConsumeContext = context;
        await HandleAsync(context.Message);
    }

    public abstract Task HandleAsync(TIntegrationEvent @event);
}

...

public sealed class UserCreatedIntegrationEventHandler : IntegrationEventHandler<UserCreatedIntegrationEvent>
{
    public override async Task HandleAsync(UserCreatedIntegrationEvent @event)
    {
        throw new System.NotImplementedException();
    }
}

The problem is that, even if I copied the event defintion, so it's exactly the same, the HandleAsync method in corresponding handler is not invoked. But when i tried to refer the User.Application.UserCreatedIntegrationEvent directly from the other service, the method is invoked correctly, but i do not really like this solution. How to solve the problem properly? Should i duplicate the definitions across other services (how to connect them using MassTransit), maybe i should move contracts to some other packages and refer the packages from both services? The code responsible for the broker dependencies registration:

internal static IServiceCollection AddRabbitMQ(this IServiceCollection services,
    IConfiguration configuration,
    bool useHealthCheck,
    Assembly consumersAssembly)
{
    var settingsSection = configuration.GetSection(RabbitMQSettingsSectionKey);
    var rabbitMQSettings = settingsSection.Get<RabbitMQSettings>();

    services
        .AddMassTransit(configurator =>
        {
            configurator.AddConsumers(consumersAssembly);
            configurator.SetKebabCaseEndpointNameFormatter();
            configurator.UsingRabbitMq((context, busFactoryConfigurator) =>
            {
                busFactoryConfigurator
                    .Host(rabbitMQSettings.HostName,
                        rabbitMQSettings.VirtualHostName,
                        hostConfigurator =>
                        {
                            hostConfigurator.Username(rabbitMQSettings.UserName);
                            hostConfigurator.Password(rabbitMQSettings.Password);
                        });

                busFactoryConfigurator.ConfigureEndpoints(context);
            });
        })
        .AddMassTransitHostedService()
        .Configure<RabbitMQSettings>(settingsSection)
        .AddScoped<IIntegrationEventPublisher, EventBus>();

    if (useHealthCheck)
    {
        services
            .AddHealthChecks()
            .AddRabbitMQ(string.Format(RabbitMQConnectionStringPattern,
                    rabbitMQSettings.HostName),
                name: RabbitConnectionCheckName,
                tags: new[] { DefaultRabbitMQTag });
    }

    return services;
}

Thanks for any help!

Upvotes: 1

Views: 1797

Answers (2)

Alexey Zimarev
Alexey Zimarev

Reputation: 19640

From the docs:

Important

MassTransit uses the full type name, including the namespace, for message contracts. When creating the same message type in two separate projects, the namespaces must match or the message will not be consumed.

The common issue is that although the contract matches by name and all the properties, the namespace is different. The consumer will try to bind to an exchange, which includes the namespace.

You can easily validate that in the RMQ management UI, by looking at the consumer queue and the bindings it has. Most probably you have two different exchanges: one, to which all the published messages go (and disappear), and the other one, which the consumer queue is bound to, which remains idle.

Upvotes: 1

Chris Patterson
Chris Patterson

Reputation: 33388

In the case where you copied the message class from the publisher to the consumer, you need to make sure the message type is the same, including namespace. This is clearly highlighted in the documentation. If the types don't match, that would explain why it wasn't consumed by the service.

Either approach, copying the files or having a shared NuGet package with the contracts, is fine. Both are widely used. MassTransit has guidelines on evolving message contracts to ensure backwards compatibility as well.

Upvotes: 3

Related Questions