Versions
- MassTransit 7.0.4
- MassTransit.RabbitMQ 7.0.4
- 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?