RabbitMQ Topology

cfg.Message<IntegrationEvent>(m => m.SetEntityName(rabbitOptions.ExchangeName));
cfg.Publish<IntegrationEvent>(p => p.ExchangeType = rabbitOptions.ExchangeType);

cfg.ReceiveEndpoint(eventBusOptions.SubscriptionClientName, e =>
{
    e.ConcurrentMessageLimit = eventBusOptions.ConcurrentMessageLimit;   // paralel işleme limiti
    e.PrefetchCount          = eventBusOptions.ConcurrentMessageLimit;   // aynı değer: dağıtım dengeli

    // ... retry / redelivery ...

    var uniqueEventTypes = subscriptionInfo.EventTypes.Values.Distinct();
    foreach (var eventEntry in uniqueEventTypes)
    {
        e.Bind(rabbitOptions.ExchangeName, s =>
        {
            s.RoutingKey   = $"{rabbitOptions.RoutingKeyPrefix}.{eventEntry}";
            s.ExchangeType = rabbitOptions.ExchangeType;
        });

        var consumerType = typeof(IntegrationEventConsumer<>).MakeGenericType(eventEntry);
        e.Consumer(consumerType, type => context.CreateScope().ServiceProvider.GetRequiredService(type));
    }
});

// Dış katman: mesaj kuyruktan ayrılır, delayed exchange üzerinden gecikmeli geri konur.
if (rabbitOptions.RedeliveryIntervalsMinutes is { Length: > 0 } redeliveryMinutes)
{
    e.UseDelayedRedelivery(r => r.Intervals(
        redeliveryMinutes.Select(min => TimeSpan.FromMinutes(min)).ToArray()));
}

// İç katman: her redelivery'de in-process kısa retry.
e.UseMessageRetry(r => r.Interval(
    rabbitOptions.RetryCount,                               // 3
    TimeSpan.FromSeconds(rabbitOptions.RetryIntervalSeconds))); // 2s

services.AddOptions<MassTransitHostOptions>().Configure(o =>
{
    o.WaitUntilStarted = false;          // RabbitMQ down ise crash etme, arka planda reconnect dene
    o.StartTimeout     = TimeSpan.Zero;  // bağlantı denemesi beklenmeden geçilir
    o.StopTimeout      = TimeSpan.FromSeconds(30);
});

{
  "MassTransitRabbitMq": {
    "Host": "rabbitmq",
    "VirtualHost": "/",
    "Username": "guest",
    "Password": "guest",
    "ExchangeName": "integration_event_bus",
    "ExchangeType": "topic",
    "RoutingKeyPrefix": "integration.event",
    "RetryCount": 3,
    "RetryIntervalSeconds": 2,
    "RedeliveryIntervalsMinutes": [0.16, 1, 5]
  },
  "EventBus": {
    "SubscriptionClientName": "DiyanetCleanArchitecture",
    "ConcurrentMessageLimit": 10
  }
}