Wolverine servicebus: динамически создавать арендаторов ⇐ C#

[Anonymous]

Недавно мы решили перейти с MassTransit на Wolverine для отправки сообщений с помощью служебной шины Azure. Однако мы столкнулись с некоторыми проблемами при нашей мультитенантной настройке.
Наша настройка выглядит следующим образом:
Кенанты создаются динамически во время выполнения, мы не знаем все идентификаторы арендаторов при запуске приложения.
Основное приложение отвечает за настройку и подключение новых арендаторов (новый контейнер не разворачивается; при подключении просто активируется приложение для Tenant).
Для каждого арендатора запускается новый контейнер-раннер и создается тема служебной шины (в Azure Service Bus) с прослушивателем для контейнера этого арендатора.
Из основного приложения нам нужно отправить сообщение в специфичную для клиента, динамически создаваемую тему.
Однако мы не можем понять, как настроить Wolverine для обработки этого, поскольку арендатор Iss (и, следовательно, имена тем) не известны при запуске.
Есть ли в Wolverine способ динамически определять или настраивать место назначения во время выполнения, чтобы сообщения можно было отправлять в соответствующую тему клиента?
Наши последние попытки выглядят следующим образом.
При использовании toTopic:

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

public async Task SendRequestToServiceBusAsync(ServiceBusRequestTopic serviceBusRequestTopic, CancellationToken cancellationToken)
{
// Create a new DeliveryOptions instance
DeliveryOptions deliveryOptions = new();

// Add headers to the existing Headers dictionary
deliveryOptions.Headers.Add("runnerId", "runner-" + serviceBusRequestTopic.RunnerId);

deliveryOptions.ToTopic("tenant-" + serviceBusRequestTopic.TenantId);

// Send the message to the topic using Wolverine
await _messageBus.SendAsync(serviceBusRequestTopic, deliveryOptions);

return true;
}

Сообщения создаются по теме по умолчанию, заданной в методе UseWolverine di (tenant-xxx):

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

builder.UseWolverine(opts =>
{
opts.UseAzureServiceBus(serviceBusOptions.SyncConnectionstring)
// Let Wolverine try to initialize any missing queues
// on the first usage at runtime
.AutoProvision()
.SystemQueuesAreEnabled(false);

// Configure message retry policy for transient failures only
opts.Policies.OnException()
.Or()
.Or() // Service Bus specific transient errors
.RetryTimes(3)
.DelayedFor(5.Seconds());

opts.PublishMessage()
.ToAzureServiceBusTopic("tenant-xxx");
}

При использовании трансляции:

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

public async Task SendRequestToServiceBusAsync(ServiceBusRequestTopic serviceBusRequestTopic, CancellationToken cancellationToken)
{
// Create a new DeliveryOptions instance
DeliveryOptions deliveryOptions = new();

// Add headers to the existing Headers dictionary
deliveryOptions.Headers.Add("runnerId", "runner-" + serviceBusRequestTopic.RunnerId);

await _messageBus.BroadcastToTopicAsync("tenant-" + serviceBusRequestTopic.TenantId, serviceBusRequestTopic, deliveryOptions);

return true;
}

Мы получаем сообщение об ошибке «Нет зарегистрированных конечных точек маршрутизации темы» из приведенного ниже фрагмента кода, скорее всего, потому, что тема не зарегистрирована в методе UseWolverine.

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

public Envelope[] RouteToTopic(T message, string topicName, DeliveryOptions? options)
{
if (message == null)
{
throw new ArgumentNullException(nameof(message));
}

if (_topicRoutes.Length == 0)
{
throw new InvalidOperationException("There are no registered topic routed endpoints");
}

var envelopes = new Envelope[_topicRoutes.Length];
for (var i = 0; i < envelopes.Length; i++)
{
envelopes[i] = _topicRoutes[i].CreateForSending(message, options, LocalDurableQueue, Runtime, topicName);
}

return envelopes;
}

Будем благодарны за любые рекомендации и примеры!

Подробнее здесь: https://stackoverflow.com/questions/797 ... te-tenants