У меня есть BackgroundService который слушает Queue.
и отправляю в Queue сообшение :
{
"Name": "ASD"
}
но когда я запускаю приложение вызывается метод StartAsync , но StartAsync не вызывает DoWork, и приложение заканчивает работу и удаляет сообшение с Queue.
Как вызвать метод DoWork?
public class Worker : BackgroundService
{
private readonly ILogger<Worker> _logger;
private readonly IBulkImporterProcessWorker<BulkQueueModel> _worker;
private readonly IMessageHandler<BulkQueueModel> _messageHandler;
private readonly IConfiguration _configuration;
private IEnumerable<IConsumer> _consumers;
private readonly StepConfiguration _steps;
public Worker(ILogger<Worker> logger, IBulkImporterProcessWorker<BulkQueueModel> worker, IMessageHandler<BulkQueueModel> messageHandler, IConfiguration configuration, IEnumerable<IConsumer> consumers)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_worker = worker ?? throw new ArgumentNullException(nameof(worker));
_messageHandler = messageHandler ?? throw new ArgumentNullException(nameof(messageHandler));
_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
_steps = configuration.GetSection("HandlerConfiguration").Get<StepConfiguration>();
_consumers = consumers ?? throw new ArgumentNullException(nameof(consumers));
}
public override Task StartAsync(CancellationToken cancellationToken)
{
var config = _configuration.GetSection("rabbitMq").Get<Bondx.RabbitMQ.Extentions.ConsumerConfiguration>();
int numberOfConsumer = _configuration.GetValue<int>("numberOfConsumer");
_consumers.SubscribeAsync<QueueMessage<BulkQueueModel>>(DoWork);
_worker.Start(null, config, Bondx.RabbitMQ.Extentions.Factory.ConsumerType.Classic, numberOfConsumer, DoWork, _logger);
return Task.CompletedTask;
}
private async Task DoWork(QueueMessage<BulkQueueModel> model, IServiceProvider provider)
{
await _messageHandler.ExecuteAsync(model.Message, _steps);
}
public override Task StopAsync(CancellationToken cancellationToken)
{
_worker.Stop();
return Task.CompletedTask;
}
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
return Task.CompletedTask;
}
Model
public class QueueModel
{
[DbParameter("id", 1)]
public string Name { get; set; }
}
**Program.CS**
public static void Main(string[] args)
{
CreateHostBuilder(args).Build().Run();
}
public static IHostBuilder CreateHostBuilder(string[] args) =>
Host.CreateDefaultBuilder(args)
.UseWindowsService()
.ConfigureLogging((ctx, logging) =>
{
logging.AddConfiguration(ctx.Configuration);
logging.AddConsole();
})
.UseNLog()
.ConfigureServices((hostContext, services) =>
{
services.AddProducer(hostContext.Configuration, "rabbitMq:Configuration");
services.AddBulkImporterHandler();
services.AddHostedService<Worker>();
});