Why is my consumer not called when run for the .NET generic host?

Viewed 43

Versions

  1. MassTransit 7.0.4
  2. MassTransit.RabbitMQ 7.0.4
  3. MassTransit.Extensions.DependencyInjection 7.0.4

I have created a Masstransit publisher and consumer. Similar to the example on the Masstransit video. However the consumer is not called.

Here the initial setup for the .NET host

namespace RedeliveryTest
{
    internal class Program
    {
        static void Main(string[] args)
        {
            CreateHostBuilder(args).Build().Run();
        }

        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureServices((hostContext, services) =>
                {
                    services.AddMassTransit(x =>
                    {

                        x.AddConsumer<RedeliveryTest.MessageConsumer>(typeof(RedeliveryTest.MessageConsumerDefinition));

                        x.SetKebabCaseEndpointNameFormatter();

                        x.UsingRabbitMq((context, cfg) =>
                        {
                            cfg.Host("rabbitmq://localhost");
                            cfg.ConfigureEndpoints(context);
                        });
                    });

                    services.AddHostedService<Worker>();
                });
    }
}

The worker publishes the contract

namespace RedeliveryTest
{
    public class Worker : BackgroundService
    {
        readonly IBus _bus;

        public Worker(IBus bus, ILogger<Worker> logger)
        {
            _bus = bus;
        }

        protected async override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                await _bus.Publish(new Message { Text = $"The time is {DateTimeOffset.Now}" }, stoppingToken);
                await Task.Delay(1000, stoppingToken);
            }
        }
    }
}

The code for consumer

namespace RedeliveryTest
{

    public class Message
    {
        public string Text { get; set; }
    }


    public class MessageConsumer :
          IConsumer<Message>
    {
        readonly ILogger<MessageConsumer> _logger;

        public MessageConsumer()
        {
        }

        public Task Consume(ConsumeContext<Message> context)
        {
            return Task.CompletedTask;
        }
    }

    public class MessageConsumerDefinition :
        ConsumerDefinition<MessageConsumer>
    {

        public MessageConsumerDefinition()
        {
            EndpointName = $"test-message-queue";
        }
        protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, IConsumerConfigurator<MessageConsumer> consumerConfigurator)
        {
            endpointConfigurator.UseMessageRetry(r => r.Intervals(500, 1000));
            endpointConfigurator.UseInMemoryOutbox();
        }
    }
}

In the rabbitmq admin page (http://localhost:15672/) I can see that the publisher is active under the "exchanges" tab. But the queue "test-message-queue" is not created.

When I create a 2nd console app to host only the consumer then that consumer does receive the messages published by the first console application. Here the source for the 2nd consumer.

namespace ConsuleConsumer
{

    public class Program
    {
        public static async Task Main(string[] args)
        {

            var services = new ServiceCollection();

            services.AddMassTransit(x =>
            {

                x.AddConsumer<RedeliveryTest.MessageConsumer>(typeof(RedeliveryTest.MessageConsumerDefinition));
                x.SetKebabCaseEndpointNameFormatter();
                x.UsingRabbitMq((context, cfg) =>
                {
                    cfg.Host("rabbitmq://localhost");
                    cfg.ConfigureEndpoints(context);
                });

            });

            var serviceProvider = services.BuildServiceProvider();
            var bus = serviceProvider.GetRequiredService<IBusControl>();

            await bus.StartAsync();

            Console.WriteLine("Press any key to exit");
            await Task.Run(() => Console.ReadKey());

            await bus.StopAsync();
        }
    }

namespace RedeliveryTest
{

    public class Message
    {
        public string Text { get; set; }
    }

    public class MessageConsumer :
        IConsumer<Message>
{
    public MessageConsumer()     {    }

    public Task Consume(ConsumeContext<Message> context)
    {
        return Task.CompletedTask;
    }
}

    public class MessageConsumerDefinition :
        ConsumerDefinition<MessageConsumer>
    {

        public MessageConsumerDefinition()
        {
            EndpointName = $"test-message-queue";
        }
        protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, IConsumerConfigurator<MessageConsumer> consumerConfigurator)
        {
            endpointConfigurator.UseMessageRetry(r => r.Intervals(500, 1000));
            endpointConfigurator.UseInMemoryOutbox();
        }
    }
}

With the 2nd console app, I can now see the queue in the rabbitmq admin page.

Question Why does the consumer when hosted in the .NET generic host not seem to work?

1 Answers

Since you're using an older version of MassTransit, the hosted service is required to start the bus. See this answer, or to summarize:

Before...

services.AddHostedService<Worker>();

Include:

services.AddMassTransitHostedService();

NOTE that this is not required if using MassTransit v8.

Related