MassTransit Consumer throws "A convention for the message type {type} was not found" exception for a registered consumer

Viewed 4869

I have a simple service that accepts requests at an HTTP Endpoint. The action for the endpoint uses MassTransit to publish an event alerting consumers that an entity has been updated. My consumer for the published event then sends a sync request to my synchronization consumer to complete the unit of work.

However, when the event consumer attempts to dispatch the request, MassTransit throws an exception saying A convention for the message type {type} was not found. This would seem to imply that my consumer is not registered, but I believe it to be.

Here's my Startup registration code:

public void ConfigureServices(IServiceCollection services)
{
    // -- removed non MassTransit code
    services.AddMassTransit(x =>
    {
        x.SetKebabCaseEndpointNameFormatter();
        x.AddConsumers(typeof(Startup).Assembly);
        x.UsingRabbitMq((context, cfg) =>
        {
            var rabbitMq = Configuration.GetSection("RabbitMq");
            var url = rabbitMq.GetValue<string>("Url");
            var username = rabbitMq.GetValue<string>("Username");
            var password = rabbitMq.GetValue<string>("Password");
            
            cfg.Host(url, h =>
            {
                h.Username(username);
                h.Password(password);
            });
            cfg.ConfigureEndpoints(context);
        });
    });
    services.AddMassTransitHostedService();
}

My API controller looks like this:

[Route("~/api/[controller]")]
public class JobSchedulingController : ApiControllerBase
{
    private readonly IPublishEndpoint _publishEndpoint;

    public JobSchedulingController(IPublishEndpoint publishEndpoint)
    {
        _publishEndpoint = publishEndpoint;
    }

    [HttpPost]
    public async Task<IActionResult> UpdateJob(JobSchedulingInputModel model)
    {
        await _publishEndpoint.Publish<JobSchedulerJobUpdated>(model);
        return Ok();
    }
}

Here's the event consumer:

public class JobSchedulerJobUpdatedConsumer 
    : IConsumer<JobSchedulerJobUpdated>
{
    public async Task Consume(
        ConsumeContext<JobSchedulerJobUpdated> context)
    {
        await context.Send<SyncJobSchedulingToLacrm>(context.Message);
    }
}

And finally, the synchronization request consumer:

public class SyncJobSchedulingToLacrmConsumer 
    : IConsumer<SyncJobSchedulingToLacrm>
{
    private readonly LacrmClient _client;
    private readonly JobSchedulingContext _context;

    public SyncJobSchedulingToLacrmConsumer(
        LacrmClient client, 
        JobSchedulingContext context)
    {
        _client = client;
        _context = context;
    }

    public async Task Consume(ConsumeContext<SyncJobSchedulingToLacrm> context)
    {
        await context.Publish<JobSchedulerSyncedToLacrm>(new
        {
            LacrmPipelineItemId = ""
        });
    }
}

I receive the error from within the event consumer and it never reaches the synchronization consumer. What could be causing this behavior?

2 Answers

You're calling Send which is a convenience method.

     await context.Send<SyncJobSchedulingToLacrm>(context.Message);

If you haven't configured an EndpointConvention for that message type, it won't find one.

I'd suggest reading this answer: https://stackoverflow.com/a/62714778/1882

I am not sure about within IConsumer, but generally you can get IBus in constructor (using dependency injection) and call GetPublishSendEndpoint<T>() method.

var sendEndpoint = await bus.GetPublishSendEndpoint<IMessage>();
await sendEndpoint.SendBatch<IMessage>(messages);
await sendEndpoint.Send<IMessage>(message);
Related