MassTransit, создайте IPublishEndpoint с заданной областью с помощью IServiceScopeFactory не публикует сообщение ⇐ C#

[Anonymous]

Моя цель — опубликовать сообщение аудита на RabbitMQ, чтобы его мог использовать другой сервис. Я фиксирую аудит внутри класса DbContext, переопределяя метод SaveChangesAsync. Моей первой мыслью было предоставить IPublishEndpoint посредством внедрения зависимостей в класс DbContext, но это не сработало. Когда я запускаю проект в Dotnet, он зависает при запуске без каких-либо ошибок, и через несколько часов я обнаружил, что это связано с UseBusOutBox().

Код: Выделить всё

 x.AddEntityFrameworkOutbox(o =>
{
o.QueryDelay = TimeSpan.FromSeconds(10);

o.UseSqlServer();
o.UseBusOutbox();
});

Затем я решил предоставить интерфейс IPublishEndpoint с помощью IServiceScopeFactory, и сначала кажется, что это удалось, поскольку служба запускается и публикует сообщение без ошибок, но затем я понял, что сообщение никогда не публиковалось. . Он даже не записывал данные в таблицы исходящих сообщений в sql.
Итак, я попробовал протестировать эту ситуацию в более простом сценарии. Я добавил код, который использовал для публикации сообщения аудита, в метод действия в контроллере. Я протестировал его, но сообщение все равно не было опубликовано. Если я внедряю IPublishEndpoint в контроллер с помощью внедрения зависимостей, публикация работает отлично.
Где-то мне не хватает конфигурации или невозможно предоставить IPublishEndpoint через IServiceScopeFactory. Буду очень рад, если кто-нибудь поможет.
Вот мой код:
Метод контроллера

Код: Выделить всё

        [Authorize]
[HttpPost()]
public async Task Create(SupplierDto model)
{

await using var scope = _serviceScopeFactory.CreateAsyncScope();

var pe = scope.ServiceProvider.GetService();
await pe.Publish(new AuditContract
{
Action = "test",
ChangedColumns = "test",
Changes = "test",
EntityName = "test",
NewValues = "test",
OldValues = "test",
PrimaryKey = "test",
TimeStamp = DateTime.Now
});
await _context.SaveChangesAsync();
return Ok();
}

Program.cs (все настройки общественного транспорта)

Код: Выделить всё

builder.Services.AddMassTransit(x =>
{
x.AddEntityFrameworkOutbox(o =>
{
o.QueryDelay = TimeSpan.FromSeconds(10);

o.UseSqlServer();
o.UseBusOutbox();
});
//Consumers
x.AddConsumersFromNamespaceContaining();
//Prefix for queues
x.SetEndpointNameFormatter(new KebabCaseEndpointNameFormatter("setup", false));

x.UsingRabbitMq((context, cfg) =>
{
cfg.ReceiveEndpoint("setup-tenant-created", e =>
{
e.UseMessageRetry(r => r.Interval(5, 5));
e.ConfigureConsumer(context);
});

cfg.Host(builder.Configuration["RabbitMq:Host"], "/", host =>
{
host.Username(builder.Configuration.GetValue("RabbitMq:Username", "guest"));
host.Password(builder.Configuration.GetValue("RabbitMq:Password", "guest"));
});

cfg.ConfigureEndpoints(context);
});
});

ОБНОВЛЕНИЕ — 18 декабря 2024 г.
Итак, проблема в UseBusOutBox. Если я уберу строку o.UseBusOutBox(); сообщение фактически публикуется, и мой потребитель получает сообщение. Таким образом, исходящий ящик Bus больше не перехватывает сообщение, а публикует его непосредственно брокеру сообщений:

Код: Выделить всё

 x.AddEntityFrameworkOutbox(o =>
{
o.QueryDelay = TimeSpan.FromSeconds(10);

o.UseSqlServer();
//o.UseBusOutbox();
});

Проблема в том, что нам нужно использовать функцию «Исходящие». Я не уверен, это ошибка или сделано специально. Дело в том, что вы не можете использовать IServiceScopeFactory для получения службы с заданной областью IPublishEndPoint, если вы используете функцию исходящих сообщений. Если мы не можем сделать что-то еще, чтобы это заработало?

Подробнее здесь: https://stackoverflow.com/questions/792 ... does-not-p