Массовый транзит: исходящие сообщения EF Core не сохраняют сообщения ⇐ C#

[Anonymous]

Я некоторое время гуглил эту проблему и, похоже, не нашел решения, похоже, это соответствует документации по массовому транспорту. Массовый транзит отлично работает в моем проекте, но как только я добавляю исходящий ящик SQL-сервера, он ничего не записывает в таблицы и не происходит никакой публикации. Вот как я регистрирую общественный транспорт:

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

_services.AddMassTransit(busConfigurator =>
{
busConfigurator.SetKebabCaseEndpointNameFormatter();
busConfigurator.AddConsumersFromNamespaceContaining();

busConfigurator.AddEntityFrameworkOutbox(options =>
{
options.UseSqlServer();
options.UseBusOutbox();
});

busConfigurator.AddConfigureEndpointsCallback((IRegistrationContext context, string name, IReceiveEndpointConfigurator cfg) =>
{
cfg.UseEntityFrameworkOutbox(context);
});

busConfigurator.UsingRabbitMq((context, configurator) =>
{
var settings = context.GetRequiredService();

configurator.Host(new Uri(settings.Host), h =>
{
h.Username(settings.Username);
h.Password(settings.Password);
});

configurator.ConfigureEndpoints(context);
});
});

У меня также есть это в OnModelCreating контекста базы данных:

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

modelBuilder.AddTransactionalOutboxEntities();

Все мои публикации осуществляются с помощью перехватчика EF Core, который фиксирует события предметной области, создаваемые моделью:

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

public override async ValueTask SavedChangesAsync(SaveChangesCompletedEventData eventData, int result, CancellationToken cancellationToken = default)
{
var bus = _serviceScope.ServiceProvider.GetRequiredService();

var context = eventData.Context;

if (context is null)
return await base.SavedChangesAsync(eventData, result, cancellationToken);

var events = context.ChangeTracker
.Entries()
.Select(a => a.Entity)
.Where(e => e.GetDomainEvents() is not null)
.SelectMany(e =>
{
var domainEvents = e.GetDomainEvents();

e.ClearDomainEvents();

return domainEvents;
});

foreach(var domainEvent in events)
await bus.PublishAsync(domainEvent, cancellationToken);

return await base.SavedChangesAsync(eventData, result, cancellationToken);
}

Наконец, моя реализация IEventBus выглядит так:

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

public async Task PublishAsync(object message, CancellationToken cancellationToken = default)
{
_logger.LogDebug("Publishing event {@Event}", message);

await _endpoint.Publish(message, (context) =>
{
context.Headers.Set("jwtToken", _contextAccessor.GetJwtTokenFromAuthHeader());
}, cancellationToken);
}

_endpoint — это внедренная точка IPublishEndpoint в этом классе.
Опять же, все работает отлично без исходящих сообщений. Что я делаю не так?

Подробнее здесь: https://stackoverflow.com/questions/788 ... g-messages