RabbitMQ как шина данных в интеграционных решениях на C# (с MassTransit)

var factory = new ConnectionFactory() { HostName = "localhost" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
 
channel.QueueDeclare(queue: "hello",
                     durable: false,
                     exclusive: false,
                     autoDelete: false,
                     arguments: null);
 
var message = "Hello World!";
var body = Encoding.UTF8.GetBytes(message);
 
channel.BasicPublish(exchange: "",
                     routingKey: "hello",
                     basicProperties: null,
                     body: body);

var factory = new ConnectionFactory
{
    HostName = "rabbit1.example.com",
    VirtualHost = "/",
    AutomaticRecoveryEnabled = true,
    NetworkRecoveryInterval = TimeSpan.FromSeconds(10)
};
 
// Альтернативные хосты для отказоустойчивого подключения
factory.SetConnectionNameList(new List<string> {
    "rabbit1.example.com",
    "rabbit2.example.com",
    "rabbit3.example.com"
});

channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
 
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (sender, ea) =>
{
    try
    {
        var body = ea.Body.ToArray();
        ProcessMessage(body); // Обработка сообщения
        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
    }
    catch
    {
        // В случае ошибки возвращаем сообщение в очередь
        channel.BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: true);
    }
};
 
channel.BasicConsume(queue: "orders",
                     autoAck: false, // Отключаем автоматические подтверждения
                     consumer: consumer);

var args = new Dictionary<string, object>
{
    {"x-dead-letter-exchange", "dead-letter-exchange"},
    {"x-dead-letter-routing-key", "dead-orders"}
};
 
channel.QueueDeclare("orders", durable: true, exclusive: false, autoDelete: false, arguments: args);

// Подключение к конкретному виртуальному хосту
var factory = new ConnectionFactory
{
    HostName = "rabbit-server",
    VirtualHost = "/production",
    UserName = "prod-app",
    Password = "very-secure-password"
};

var properties = channel.CreateBasicProperties();
properties.Persistent = true; // Сообщение будет сохранено на диск
 
channel.BasicPublish(exchange: "orders",
                     routingKey: "new",
                     basicProperties: properties,
                     body: largeMessageBody);

// Издатель
var fileReference = new FileReference
{
    FileId = Guid.NewGuid(),
    StoragePath = $"s3://company-bucket/exports/{fileName}",
    ContentType = "application/pdf",
    CreatedAt = DateTime.UtcNow
};
 
await s3Client.UploadFileAsync(localFilePath, fileReference.StoragePath);
 
var message = JsonSerializer.Serialize(fileReference);
var body = Encoding.UTF8.GetBytes(message);
 
channel.BasicPublish(exchange: "files",
                     routingKey: "processed",
                     basicProperties: null,
                     body: body);

// Создание прямого обменника
channel.ExchangeDeclare(
    exchange: "orders", 
    type: ExchangeType.Direct);
 
// Привязка очереди к обменнику с ключом маршрутизации
channel.QueueBind(
    queue: "fulfillment_orders", 
    exchange: "orders", 
    routingKey: "new");
 
// Отправка сообщения с конкретным ключом маршрутизации
channel.BasicPublish(
    exchange: "orders",
    routingKey: "new",
    basicProperties: null,
    body: Encoding.UTF8.GetBytes(message));

channel.ExchangeDeclare(
    exchange: "logs", 
    type: ExchangeType.Topic);
 
// Разные системы подписываются на разные паттерны сообщений
channel.QueueBind(
    queue: "critical_alerts", 
    exchange: "logs", 
    routingKey: "*.critical.*");
 
channel.QueueBind(
    queue: "all_errors", 
    exchange: "logs", 
    routingKey: "error.#");
 
// Отправка сообщения по определённому пути
channel.BasicPublish(
    exchange: "logs",
    routingKey: "error.critical.payment",
    basicProperties: null,
    body: logMessage);

channel.ExchangeDeclare(
    exchange: "broadcasts", 
    type: ExchangeType.Fanout);
 
// Разные подсистемы подписываются на все сообщения
channel.QueueBind(
    queue: "audit_log", 
    exchange: "broadcasts", 
    routingKey: ""); // Ключ игнорируется
 
channel.QueueBind(
    queue: "notifications", 
    exchange: "broadcasts", 
    routingKey: "");
 
// Любое опубликованное сообщение попадёт во все очереди
channel.BasicPublish(
    exchange: "broadcasts",
    routingKey: "", // Можно указать любой ключ
    basicProperties: null,
    body: broadcastMessage);

channel.ExchangeDeclare(
    exchange: "filtered_data", 
    type: ExchangeType.Headers);
 
// Можно настроить сопоставление по заголовкам
var bindingArgs = new Dictionary<string, object>
{
    {"format", "pdf"},
    {"type", "report"},
    {"x-match", "all"} // Требуется совпадение всех указанных заголовков
};
 
channel.QueueBind(
    queue: "pdf_reports", 
    exchange: "filtered_data", 
    routingKey: "",
    arguments: bindingArgs);
 
// При публикации указываем нужные заголовки
var props = channel.CreateBasicProperties();
props.Headers = new Dictionary<string, object>
{
    {"format", "pdf"},
    {"type", "report"},
    {"department", "finance"}
};
 
channel.BasicPublish(
    exchange: "filtered_data",
    routingKey: "",
    basicProperties: props,
    body: reportData);

public class SapAdapter : IHostedService
{
    private readonly IBus _bus;
    private readonly ISapClient _sapClient;
    private readonly ILogger<SapAdapter> _logger;
 
    public SapAdapter(IBus bus, ISapClient sapClient, ILogger<SapAdapter> logger)
    {
        _bus = bus;
        _sapClient = sapClient;
        _logger = logger;
    }
 
    public Task StartAsync(CancellationToken cancellationToken)
    {
        _bus.SubscribeHandler<CreateOrderCommand>(HandleCreateOrder);
        _bus.SubscribeHandler<UpdateInventoryCommand>(HandleUpdateInventory);
        // Подписки на другие команды
        
        return Task.CompletedTask;
    }
 
    private async Task HandleCreateOrder(CreateOrderCommand command)
    {
        try
        {
            var sapOrder = TranslateToSapOrder(command);
            var result = await _sapClient.CreateOrderAsync(sapOrder);
            
            // Публикуем событие о результате создания заказа
            await _bus.Publish(new OrderCreatedInErpEvent
            {
                OrderId = command.OrderId,
                ErpOrderId = result.ErpOrderId,
                Timestamp = DateTime.UtcNow
            });
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка при создании заказа в SAP");
            // Публикуем событие об ошибке
            await _bus.Publish(new OrderCreationFailedEvent
            {
                OrderId = command.OrderId,
                ErrorMessage = ex.Message
            });
        }
    }
    
    // Другие обработчики
}

public class MappingProfiles : Profile
{
    public MappingProfiles()
    {
        CreateMap<DynamicsOrder, SapOrder>()
            .ForMember(dest => dest.VBELN, opt => opt.MapFrom(src => src.OrderNumber))
            .ForMember(dest => dest.ERDAT, opt => opt.MapFrom(src => 
                ConvertToSapDate(src.CreatedOn)))
            .AfterMap((src, dest) => ValidateSapOrder(dest));
            
        CreateMap<DynamicsOrder, ProductionOrder>()
            .ForMember(dest => dest.Priority, opt => opt.MapFrom(src => 
                MapPriorityFromDynamics(src.PriorityCode)))
            .AfterMap((src, dest) => CalculateProductionSlots(dest));
            
        // Другие маппинги
    }
    
    private string ConvertToSapDate(DateTime date)
    {
        // Преобразование в специфичный для SAP формат даты
        return date.ToString("yyyyMMdd");
    }
    
    private void ValidateSapOrder(SapOrder order)
    {
        // Дополнительная валидация перед отправкой в SAP
        if (string.IsNullOrEmpty(order.VBELN))
            throw new ValidationException("Номер заказа не может быть пустым для SAP");
    }
    
    // Другие вспомогательные методы
}

public class OrderProcessingSaga : MassTransitStateMachine<OrderProcessingSagaState>
{
    public OrderProcessingSaga()
    {
        InstanceState(x => x.CurrentState);
        
        Event(() => OrderSubmitted, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => ErpOrderCreated, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => ProductionScheduled, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => InventoryAllocated, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => OrderFailed, x => x.CorrelateById(m => m.Message.OrderId));
        
        Initially(
            When(OrderSubmitted)
                .Then(ctx => { ctx.Instance.OrderDetails = ctx.Data; })
                .TransitionTo(AwaitingErpProcessing)
                .Publish(ctx => new CreateErpOrderCommand(ctx.Instance)));
        
        During(AwaitingErpProcessing,
            When(ErpOrderCreated)
                .TransitionTo(AwaitingProductionScheduling)
                .Publish(ctx => new ScheduleProductionCommand(ctx.Instance)),
            When(OrderFailed)
                .TransitionTo(FailedState)
                .Publish(ctx => new NotifyOrderFailureCommand(ctx.Instance)));
        
        // Другие состояния и переходы
    }
    
    public State AwaitingErpProcessing { get; private set; }
    public State AwaitingProductionScheduling { get; private set; }
    public State AwaitingInventoryAllocation { get; private set; }
    public State OrderCompleted { get; private set; }
    public State FailedState { get; private set; }
    
    public Event<OrderSubmittedEvent> OrderSubmitted { get; private set; }
    public Event<ErpOrderCreatedEvent> ErpOrderCreated { get; private set; }
    // Другие события
}

During(AwaitingProductionScheduling,
    When(ProductionSchedulingFailed)
        .Then(ctx => {
            _logger.LogWarning($"Не удалось запланировать производство для заказа {ctx.Instance.OrderId}");
        })
        .Publish(ctx => new CancelErpOrderCommand { OrderId = ctx.Instance.OrderId })
        .TransitionTo(CompensatingTransactions));

var policy = Policy
    .RateLimitAsync(10, TimeSpan.FromSeconds(1)) // Не более 10 запросов в секунду
    .WrapAsync(Policy
    .BulkheadAsync(5)); // Не более 5 одновременных запросов
 
await policy.ExecuteAsync(async () => await _sapClient.CreateOrderAsync(sapOrder));

// Настройка обменников для потока заказов
channel.ExchangeDeclare("orders.commands", ExchangeType.Direct);
channel.ExchangeDeclare("orders.events", ExchangeType.Topic);

// Подписка производственной системы на события заказов
channel.QueueBind(
    queue: "production.order_events",
    exchange: "orders.events",
    routingKey: "order.created.*");
    
channel.QueueBind(
    queue: "production.order_events",
    exchange: "orders.events",
    routingKey: "order.updated.priority");

// Пример структуры микросервиса-адаптера
public class Program
{
    public static void Main(string[] args)
    {
        CreateHostBuilder(args).Build().Run();
    }
 
    public static IHostBuilder CreateHostBuilder(string[] args) =>
        Host.CreateDefaultBuilder(args)
            .ConfigureServices((hostContext, services) =>
            {
                // Регистрация MassTransit с настройкой RabbitMQ
                services.AddMassTransit(x =>
                {
                    // Регистрация обработчиков сообщений
                    x.AddConsumer<CreateOrderInSapConsumer>();
                    x.AddConsumer<UpdateProductionScheduleConsumer>();
                    
                    x.UsingRabbitMq((context, cfg) =>
                    {
                        cfg.Host(hostContext.Configuration.GetConnectionString("RabbitMQ"));
                        
                        // Настройка приёма команд
                        cfg.ReceiveEndpoint("erp-orders-commands", e =>
                        {
                            e.ConfigureConsumer<CreateOrderInSapConsumer>(context);
                            // Настройка повторных попыток
                            e.UseMessageRetry(r => r.Interval(3, TimeSpan.FromSeconds(5)));
                            // Настройка ограничения конкурентности
                            e.PrefetchCount = 4;
                        });
                        
                        // Другие эндпоинты
                    });
                });
                
                // Регистрация других зависимостей
                services.AddSingleton<ISapClient, SapClient>();
                services.AddHostedService<SapAdapterService>();
            });
}

// Контрактная библиотека (Messages.dll)
namespace TechnoProm.Integration.Messages
{
    // События от CRM
    public interface IOrderCreated
    {
        Guid OrderId { get; }
        string CustomerCode { get; }
        DateTime OrderDate { get; }
        List<OrderLineItem> Lines { get; }
        OrderPriority Priority { get; }
    }
    
    // Команды для ERP
    public interface ICreateErpOrder
    {
        Guid OrderId { get; }
        string CustomerCode { get; }
        DateTime OrderDate { get; }
        List<OrderLineItem> Lines { get; }
    }
    
    // События от ERP
    public interface IErpOrderCreated
    {
        Guid OrderId { get; }
        string ErpOrderNumber { get; }
        DateTime ProcessedDate { get; }
    }
    
    // Модели данных
    public class OrderLineItem
    {
        public string ProductCode { get; set; }
        public int Quantity { get; set; }
        public decimal UnitPrice { get; set; }
    }
    
    public enum OrderPriority
    {
        Normal,
        High,
        Urgent
    }
}

// Пример обработчика команды для создания заказа в SAP
public class CreateOrderInSapConsumer : IConsumer<ICreateErpOrder>
{
    private readonly ISapClient _sapClient;
    private readonly ILogger<CreateOrderInSapConsumer> _logger;
    private readonly IBus _bus;
    
    public CreateOrderInSapConsumer(ISapClient sapClient, ILogger<CreateOrderInSapConsumer> logger, IBus bus)
    {
        _sapClient = sapClient;
        _logger = logger;
        _bus = bus;
    }
    
    public async Task Consume(ConsumeContext<ICreateErpOrder> context)
    {
        _logger.LogInformation($"Получена команда на создание заказа {context.Message.OrderId} в SAP");
        
        try
        {
            // Преобразование в формат SAP
            var sapOrder = new SapOrderCreate
            {
                KUNNR = context.Message.CustomerCode,
                AUDAT = context.Message.OrderDate.ToString("yyyyMMdd"),
                ITEMS = context.Message.Lines.Select(l => new SapOrderItem
                {
                    MATNR = l.ProductCode,
                    KWMENG = l.Quantity.ToString(),
                    NETPR = l.UnitPrice.ToString("F2", CultureInfo.InvariantCulture)
                }).ToList()
            };
            
            // Вызов SAP API
            var sapResponse = await _sapClient.CreateOrderAsync(sapOrder);
            
            if (sapResponse.Success)
            {
                // Публикация события об успешном создании заказа в ERP
                await _bus.Publish<IErpOrderCreated>(new
                {
                    OrderId = context.Message.OrderId,
                    ErpOrderNumber = sapResponse.OrderNumber,
                    ProcessedDate = DateTime.UtcNow
                });
                
                _logger.LogInformation($"Заказ {context.Message.OrderId} успешно создан в SAP с номером {sapResponse.OrderNumber}");
            }
            else
            {
                // Публикация события об ошибке
                await _bus.Publish<IErpOrderRejected>(new
                {
                    OrderId = context.Message.OrderId,
                    Reason = sapResponse.ErrorMessage,
                    ProcessedDate = DateTime.UtcNow
                });
                
                _logger.LogWarning($"SAP отклонил заказ {context.Message.OrderId}: {sapResponse.ErrorMessage}");
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, $"Ошибка при создании заказа {context.Message.OrderId} в SAP");
            
            // В случае исключения также публикуем событие ошибки
            await _bus.Publish<IErpOrderRejected>(new
            {
                OrderId = context.Message.OrderId,
                Reason = $"Внутренняя ошибка: {ex.Message}",
                ProcessedDate = DateTime.UtcNow
            });
            
            // Перебрасываем исключение, чтобы активировать политику повторных попыток
            throw;
        }
    }
}

// Состояние саги для отслеживания процесса обработки заказа
public class OrderProcessingState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; }
    
    // Данные для отслеживания процесса
    public Guid OrderId { get; set; }
    public string CustomerCode { get; set; }
    public string ErpOrderNumber { get; set; }
    public string ProductionBatchId { get; set; }
    public string InventoryReservationId { get; set; }
    
    // Данные для компенсирующих действий
    public bool IsErpOrderCreated { get; set; }
    public bool IsProductionScheduled { get; set; }
    public bool IsInventoryReserved { get; set; }
    
    // Данные для отслеживания ошибок
    public string LastErrorMessage { get; set; }
    public int RetryCount { get; set; }
}
 
// Машина состояний для обработки заказа
public class OrderProcessingSaga : MassTransitStateMachine<OrderProcessingState>
{
    public OrderProcessingSaga()
    {
        // Определение событий
        Event(() => OrderCreated, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => ErpOrderCreated, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => ErpOrderRejected, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => ProductionScheduled, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => InventoryReserved, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => OrderCompletionFailed, x => x.CorrelateById(m => m.Message.OrderId));
 
        // Определение начального состояния
        Initially(
            When(OrderCreated)
                .Then(context => {
                    context.Instance.OrderId = context.Data.OrderId;
                    context.Instance.CustomerCode = context.Data.CustomerCode;
                })
                .ThenAsync(context => Console.Out.WriteLineAsync($"Начата обработка заказа {context.Instance.OrderId}"))
                .Publish(context => new CreateErpOrderCommand {
                    OrderId = context.Instance.OrderId,
                    CustomerCode = context.Instance.CustomerCode,
                    // Другие данные из события
                })
                .TransitionTo(AwaitingErpConfirmation)
        );
 
        // Другие состояния и переходы
    }
 
    // Определение состояний
    public State AwaitingErpConfirmation { get; private set; }
    public State AwaitingProductionScheduling { get; private set; }
    public State AwaitingInventoryReservation { get; private set; }
    public State Completed { get; private set; }
    public State Faulted { get; private set; }
 
    // Определение событий
    public Event<IOrderCreated> OrderCreated { get; private set; }
    public Event<IErpOrderCreated> ErpOrderCreated { get; private set; }
    public Event<IErpOrderRejected> ErpOrderRejected { get; private set; }
    // Другие события
}

// Запрос наличия товара
public async Task<bool> CheckProductAvailability(string productCode, int quantity)
{
    var requestClient = _busControl.CreateRequestClient<CheckInventoryRequest>(TimeSpan.FromSeconds(10));
    
    var response = await requestClient.GetResponse<InventoryCheckResponse>(new
    {
        ProductCode = productCode,
        Quantity = quantity,
        RequestTime = DateTime.UtcNow
    });
    
    return response.Message.IsAvailable;
}

// Публикация события обновления цен
await _bus.Publish<IPriceListUpdated>(new
{
    PriceListId = updatedPriceList.Id,
    UpdatedProducts = changedPriceItems,
    EffectiveDate = updatedPriceList.EffectiveDate,
    UpdatedBy = currentUser,
    Timestamp = DateTime.UtcNow
});

// Отправка команды на создание производственного задания
await _endpointFactory.Send<IScheduleProduction>(new
{
    OrderId = order.Id,
    ProductCode = order.ProductCode,
    Quantity = order.Quantity,
    RequiredDate = order.DeliveryDate,
    Priority = MapOrderPriorityToProductionPriority(order.Priority)
});

services.AddMassTransit(x =>
{
    x.AddConsumers(Assembly.GetExecutingAssembly());
    
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(configuration["RabbitMQ:Host"], h =>
        {
            h.Username(configuration["RabbitMQ:Username"]);
            h.Password(configuration["RabbitMQ:Password"]);
        });
        
        // Настройка System.Text.Json для всех сообщений
        cfg.ConfigureJsonSerializerOptions(options =>
        {
            options.PropertyNamingPolicy = JsonNamingPolicy.CamelCase;
            options.DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull;
            options.PropertyNameCaseInsensitive = true;
            options.Converters.Add(new JsonStringEnumConverter());
            
            return options;
        });
        
        cfg.ConfigureEndpoints(context);
    });
});

// Обработчик команды (write model)
public class UpdateInventoryConsumer : IConsumer<IUpdateInventory>
{
    private readonly IInventoryRepository _repository;
    
    public UpdateInventoryConsumer(IInventoryRepository repository)
    {
        _repository = repository;
    }
    
    public async Task Consume(ConsumeContext<IUpdateInventory> context)
    {
        var inventory = await _repository.GetByProductCode(context.Message.ProductCode);
        
        inventory.UpdateQuantity(context.Message.QuantityDelta);
        
        await _repository.Save(inventory);
        
        // Публикация события об изменении инвентаря
        await context.Publish<IInventoryChanged>(new
        {
            ProductCode = inventory.ProductCode,
            NewQuantity = inventory.AvailableQuantity,
            ChangeReason = context.Message.Reason
        });
    }
}
 
// Обработчик запроса (read model)
public class GetInventoryStatusConsumer : IConsumer<IGetInventoryStatus>
{
    private readonly IInventoryReadOnlyRepository _readRepository;
    
    public GetInventoryStatusConsumer(IInventoryReadOnlyRepository readRepository)
    {
        _readRepository = readRepository;
    }
    
    public async Task Consume(ConsumeContext<IGetInventoryStatus> context)
    {
        var status = await _readRepository.GetCurrentStatusByProductCode(context.Message.ProductCode);
        
        await context.RespondAsync<IInventoryStatus>(new
        {
            ProductCode = status.ProductCode,
            AvailableQuantity = status.Quantity,
            ReservedQuantity = status.Reserved,
            LastUpdated = status.LastUpdatedUtc
        });
    }
}

public class CachedProductCatalogService : IProductCatalogService
{
    private readonly IProductCatalogService _innerService;
    private readonly IMemoryCache _cache;
    private readonly ILogger<CachedProductCatalogService> _logger;
    
    // Константы для настройки кэширования
    private const string AllProductsCacheKey = "AllProducts";
    private const string ProductByCodeKeyPrefix = "Product_";
    private static readonly TimeSpan CacheDuration = TimeSpan.FromMinutes(15);
    
    public CachedProductCatalogService(
        IProductCatalogService innerService,
        IMemoryCache cache,
        ILogger<CachedProductCatalogService> logger)
    {
        _innerService = innerService;
        _cache = cache;
        _logger = logger;
    }
    
    public async Task<ProductDetails> GetProductByCode(string productCode)
    {
        var cacheKey = $"{ProductByCodeKeyPrefix}{productCode}";
        
        if (!_cache.TryGetValue(cacheKey, out ProductDetails product))
        {
            _logger.LogDebug($"Cache miss for product {productCode}");
            
            product = await _innerService.GetProductByCode(productCode);
            
            if (product != null)
            {
                var cacheOptions = new MemoryCacheEntryOptions()
                    .SetAbsoluteExpiration(CacheDuration)
                    .RegisterPostEvictionCallback((key, value, reason, state) =>
                    {
                        _logger.LogTrace($"Product {productCode} evicted from cache: {reason}");
                    });
                
                _cache.Set(cacheKey, product, cacheOptions);
            }
        }
        
        return product;
    }
}

public class IdempotentConsumer<TMessage> : IConsumer<TMessage> where TMessage : class
{
    private readonly IConsumer<TMessage> _innerConsumer;
    private readonly IProcessedMessagesRepository _repository;
    
    public IdempotentConsumer(IConsumer<TMessage> innerConsumer, IProcessedMessagesRepository repository)
    {
        _innerConsumer = innerConsumer;
        _repository = repository;
    }
    
    public async Task Consume(ConsumeContext<TMessage> context)
    {
        var messageId = context.MessageId?.ToString() ?? throw new InvalidOperationException("Сообщение без ID");
        
        if (await _repository.IsProcessed(messageId))
        {
            // Сообщение уже обрабатывалось - игнорируем
            return;
        }
        
        // Помечаем как "в обработке" перед выполнением
        await _repository.MarkProcessing(messageId);
        
        try
        {
            // Вызываем внутренний обработчик
            await _innerConsumer.Consume(context);
            
            // Помечаем как успешно обработанное
            await _repository.MarkProcessed(messageId);
        }
        catch (Exception)
        {
            // В случае ошибки помечаем как "не обработано"
            await _repository.MarkFailed(messageId);
            throw;
        }
    }
}

services.AddScoped<IConsumer<CreateOrderCommand>, CreateOrderConsumer>();
services.AddScoped<IConsumer<CreateOrderCommand>>(provider => 
    new IdempotentConsumer<CreateOrderCommand>(
        provider.GetRequiredService<CreateOrderConsumer>(),
        provider.GetRequiredService<IProcessedMessagesRepository>()
    ));

public static class RabbitMqBusExtensions
{
    // Упрощение отправки команд с корреляцией
    public static async Task SendWithCorrelation<T>(this IBus bus, string queueName, T message, Guid correlationId) 
        where T : class
    {
        var sendEndpoint = await bus.GetSendEndpoint(new Uri($"queue:{queueName}"));
        
        await sendEndpoint.Send(message, context => {
            context.CorrelationId = correlationId;
            context.Headers.Set("Created", DateTime.UtcNow);
        });
    }
    
    // Быстрый запрос-ответ с таймаутом
    public static async Task<TResponse> RequestWithTimeout<TRequest, TResponse>(
        this IBus bus, 
        TRequest request,
        TimeSpan timeout) 
        where TRequest : class
        where TResponse : class
    {
        var client = bus.CreateRequestClient<TRequest>(timeout);
        var response = await client.GetResponse<TResponse>(request);
        return response.Message;
    }
}

public class LargeMessageHandler
{
    private readonly IBlobStorage _storage;
    private readonly IBus _bus;
    
    // При отправке большого сообщения
    public async Task SendLargeMessage(byte[] messageData, string messageType)
    {
        // Если размер превышает порог — сохраняем в хранилище
        if (messageData.Length > MaxMessageSize)
        {
            var blobId = Guid.NewGuid().ToString();
            await _storage.Store(blobId, messageData);
            
            // Отправляем только ссылку на данные
            await _bus.Publish<ILargeMessageReference>(new 
            {
                BlobId = blobId,
                MessageType = messageType,
                Size = messageData.Length,
                ExpiresAt = DateTime.UtcNow.AddDays(1)
            });
        }
        else
        {
            // Маленькие сообщения отправляем как обычно
            await _bus.Publish(messageData, messageType);
        }
    }
}

public class LargeMessageReferenceConsumer : IConsumer<ILargeMessageReference>
{
    private readonly IBlobStorage _storage;
    private readonly IBus _bus;
    
    public async Task Consume(ConsumeContext<ILargeMessageReference> context)
    {
        var reference = context.Message;
        
        // Получаем данные из хранилища
        var data = await _storage.Retrieve(reference.BlobId);
        
        // Динамически определяем тип и выполняем десериализацию
        var messageType = Type.GetType(reference.MessageType);
        var message = JsonSerializer.Deserialize(data, messageType);
        
        // Отправляем обработчику нужного типа
        await context.Forward(message);
    }
}

// Настройка политики повторных попыток на уровне получателя
cfg.ReceiveEndpoint("erp-command-queue", e =>
{
    // Повторы с увеличивающимися интервалами
    e.UseMessageRetry(r => r.Incremental(3, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3)));
    
    // Перенаправление в очередь ошибок после исчерпания попыток
    e.UseDelayedRedelivery(r => r.Intervals(TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(15), TimeSpan.FromMinutes(30)));
    e.UseInMemoryOutbox();
    
    e.ConfigureConsumer<ErpCommandConsumer>(context);
});

public async Task ExecuteWithRetry(Func<Task> action)
{
    int attempt = 0;
    int maxAttempts = 5;
    TimeSpan delay = TimeSpan.FromSeconds(1);
    Random jitter = new Random();
    
    while (true)
    {
        try
        {
            await action();
            return; // Успех - выходим из цикла
        }
        catch (SapTemporaryException ex) // Временные ошибки SAP
        {
            attempt++;
            
            if (attempt >= maxAttempts)
                throw new MaxRetryAttemptsExceededException($"Исчерпаны все {maxAttempts} попыток", ex);
            
            // Экспоненциальная задержка со случайным компонентом для предотвращения наложений
            int jitterMilliseconds = jitter.Next(-500, 500);
            TimeSpan waitTime = delay + TimeSpan.FromMilliseconds(jitterMilliseconds);
            
            _logger.LogWarning($"Попытка {attempt} не удалась, повтор через {waitTime.TotalSeconds:F1} секунд. Ошибка: {ex.Message}");
            
            await Task.Delay(waitTime);
            
            // Увеличиваем интервал для следующей попытки
            delay = TimeSpan.FromSeconds(Math.Min(60, Math.Pow(2, attempt)));
        }
        catch (SapFatalException ex) // Неисправимые ошибки - не пытаемся повторить
        {
            _logger.LogError(ex, "Фатальная ошибка SAP, повторы невозможны");
            throw;
        }
    }
}

// Регистрация Circuit Breaker
services.AddSingleton<ICircuitBreakerPolicy>(provider => 
    new CircuitBreakerPolicy(
        failureThreshold: 0.3, // 30% сбоев
        samplingDuration: TimeSpan.FromMinutes(1),
        minimumThroughput: 10,
        breakDuration: TimeSpan.FromMinutes(2),
        provider.GetService<ILogger<CircuitBreakerPolicy>>()
    )
);

[Test]
public async Task When_order_created_should_publish_erp_command()
{
 // Настройка тестового стенда
 var harness = new InMemoryTestHarness();
 var consumerHarness = harness.Consumer<OrderCreatedConsumer>();
 
 await harness.Start();
 try
 {
     // Отправка тестового сообщения
     await harness.Bus.Publish<IOrderCreated>(new 
     {
         OrderId = Guid.NewGuid(),
         CustomerCode = "TEST001",
         OrderDate = DateTime.UtcNow,
         Lines = new List<OrderLineItem>()
     });
     
     // Проверка, что потребитель получил сообщение
     Assert.True(await consumerHarness.Consumed.Any<IOrderCreated>());
     
     // Проверка, что была опубликована ожидаемая команда
     Assert.True(await harness.Published.Any<ICreateErpOrder>());
 }
 finally
 {
     await harness.Stop();
 }
}

public class SapIntegrationTests : IClassFixture<RabbitMqFixture>
{
 private readonly RabbitMqFixture _fixture;
 private readonly MockSapService _mockSap;
 
 public SapIntegrationTests(RabbitMqFixture fixture)
 {
     _fixture = fixture;
     _mockSap = new MockSapService();
     
     // Настройка заглушки SAP с контролируемыми ответами
     _mockSap.SetupOrderCreationResponse(new SapOrderResponse { Success = true, OrderNumber = "4500001234" });
 }
 
 [Fact]
 public async Task Complete_order_flow_with_sap_integration()
 {
     // Код теста с реальным RabbitMQ и заглушкой SAP
 }
}

[Test]
public async Task System_recovers_after_rabbitmq_restart()
{
 // Отправка сообщения
 await _bus.Publish<IOrderCreated>(CreateTestOrder());
 
 // Проверка, что сообщение начало обрабатываться
 Assert.True(await WaitForMessageProcessingStarted());
 
 // Перезапуск RabbitMQ через Docker
 await _dockerControl.RestartContainer("rabbitmq_test");
 
 // Проверка, что обработка продолжилась после восстановления соединения
 Assert.True(await WaitForMessageProcessingCompleted());
}

[Benchmark]
public async Task ProcessOrdersUnderLoad()
{
 var tasks = new List<Task>();
 for (int i = 0; i < 100; i++)
 {
     var order = GenerateRandomOrder();
     tasks.Add(_bus.Publish<IOrderCreated>(order));
 }
 await Task.WhenAll(tasks);
}

public class MessageContractTests
{
 [Theory]
 [MemberData(nameof(GetMessageContracts))]
 public void Validate_message_against_schema(Type messageType, string schemaPath)
 {
     // Создание тестового сообщения
     var message = Activator.CreateInstance(messageType);
     // Заполнение тестовыми данными
     
     // Сериализация
     var json = JsonSerializer.Serialize(message);
     
     // Валидация против схемы
     var schema = JSchema.Parse(File.ReadAllText(schemaPath));
     var jObject = JObject.Parse(json);
     
     bool isValid = jObject.IsValid(schema, out IList<ValidationError> errors);
     Assert.True(isValid, string.Join(Environment.NewLine, errors.Select(e => e.Message)));
 }
}