Брокер NATS в C#

PUB subject.name 11
Hello World

using NATS.Client;
 
// Создание фабрики подключений
var connectionFactory = new ConnectionFactory();
 
// Подключение к серверу NATS
IConnection connection = connectionFactory.CreateConnection("nats://localhost:4222");
 
// Теперь можно использовать connection для отправки и получения сообщений

var options = ConnectionFactory.GetDefaultOptions();
options.Url = "nats://localhost:4222";
options.AllowReconnect = true;
options.MaxReconnect = 3;
options.ReconnectWait = 2000; // 2 секунды между попытками переподключения
options.Name = "my-csharp-app";
 
IConnection connection = new ConnectionFactory().CreateConnection(options);

// Публикация строкового сообщения
connection.Publish("greeting.general", Encoding.UTF8.GetBytes("Hello, NATS world!"));
 
// Публикация объекта через JSON-сериализацию
var order = new Order { Id = 12345, CustomerName = "John Doe", TotalAmount = 99.95m };
string json = JsonSerializer.Serialize(order);
connection.Publish("orders.new", Encoding.UTF8.GetBytes(json));

await connection.PublishAsync("greeting.async", Encoding.UTF8.GetBytes("Async hello!"));

// Простая подписка
IAsyncSubscription subscription = connection.SubscribeAsync("greeting.general", (sender, args) =>
{
    string receivedMessage = Encoding.UTF8.GetString(args.Message.Data);
    Console.WriteLine($"Получено сообщение: {receivedMessage}");
});
 
// Подписка с использованием подстановочных знаков
connection.SubscribeAsync("orders.*", (sender, args) =>
{
    string subject = args.Message.Subject; // например, "orders.new"
    string data = Encoding.UTF8.GetString(args.Message.Data);
    Console.WriteLine($"Получено сообщение по теме {subject}: {data}");
});

// Подписка с использованием группы для балансировки нагрузки
connection.SubscribeAsync("tasks.process", "workers", (sender, args) =>
{
    string task = Encoding.UTF8.GetString(args.Message.Data);
    Console.WriteLine($"Обработка задачи: {task}");
    // Обработка задачи...
});

// Асинхронная подписка с обработкой сообщений в отдельном потоке
var subscription = await connection.SubscribeAsync("orders.incoming");
var task = subscription.Start(msg =>
{
    // Этот код выполняется в пуле потоков
    var orderData = Encoding.UTF8.GetString(msg.Data);
    ProcessOrder(orderData);  // какая-то долгая операция
});
 
// Можно дождаться завершения обработки
await task;
 
// Или остановить обработку в любой момент
subscription.Unsubscribe();

// Обработка событий соединения
Options opts = ConnectionFactory.GetDefaultOptions();
opts.Url = "nats://demo.nats.io:4222";
opts.ClosedEventHandler = (sender, args) => 
{
    Console.WriteLine("Соединение закрыто: " + args.Connection.ConnectedUrl);
};
opts.DisconnectedEventHandler = (sender, args) => 
{
    Console.WriteLine("Соединение потеряно: " + args.Connection.ConnectedUrl);
};
opts.ReconnectedEventHandler = (sender, args) => 
{
    Console.WriteLine("Переподключено к: " + args.Connection.ConnectedUrl);
};
 
using (var conn = new ConnectionFactory().CreateConnection(opts))
{
    // Работа с соединением
}

// Подписка на конкретный субъект
connection.SubscribeAsync("orders.created", MessageHandler);
 
// Подписка на все события заказов с использованием "*"
connection.SubscribeAsync("orders.*", MessageHandler);
// Получит: orders.created, orders.shipped, orders.cancelled и т.д.
 
// Подписка на все события, связанные с пользователем, используя ">"
connection.SubscribeAsync("user.123.>", MessageHandler);
// Получит: user.123.profile.updated, user.123.order.created и т.д.

var subscription = connection.SubscribeAsync("heavy.workload");
subscription.MessageHandler += (sender, args) =>
{
    ProcessComplexMessage(args.Message); // длительная операция
};
 
// Устанавливаем лимит на 100 одновременно обрабатываемых сообщений
subscription.SetPendingLimits(100, 1024 * 1024 * 10); // 100 сообщений или 10MB
subscription.Start();

// Установка таймаута в 5 секунд для публикации сообщения
var publishOptions = new PublishOptions
{
    Timeout = TimeSpan.FromSeconds(5)
};
 
await connection.PublishAsync("time.sensitive.topic", data, publishOptions);

// На стороне клиента
string response = Encoding.UTF8.GetString(
    connection.Request("math.sum", Encoding.UTF8.GetBytes("40 + 2"))
);
Console.WriteLine($"Ответ: {response}"); // Выведет "42"
 
// На стороне сервера
connection.Subscribe("math.sum", (sender, args) =>
{
    string request = Encoding.UTF8.GetString(args.Message.Data);
    // Разбираем запрос и вычисляем результат
    string[] parts = request.Split('+');
    int a = int.Parse(parts[0].Trim());
    int b = int.Parse(parts[1].Trim());
    int result = a + b;
    
    // Отправляем ответ
    connection.Publish(args.Message.Reply, Encoding.UTF8.GetBytes(result.ToString()));
});

// Асинхронный запрос с таймаутом
try
{
    var reply = await connection.RequestAsync("service.action", 
                                            requestData, 
                                            timeout: TimeSpan.FromSeconds(5));
    // Обработка ответа
}
catch (NATSTimeoutException)
{
    // Обработка ситуации, когда сервис не ответил вовремя
}

// Запускаем на нескольких серверах
connection.Subscribe("database.query", "db-workers", (sender, args) =>
{
    string query = Encoding.UTF8.GetString(args.Message.Data);
    // Выполняем запрос к БД
    string results = ExecuteDatabaseQuery(query);
    
    // Отправляем результаты обратно
    connection.Publish(args.Message.Reply, Encoding.UTF8.GetBytes(results));
});

// Клиент просто отправляет запрос, не заботясь о том,
// сколько серверов обрабатывают запросы
var response = connection.Request("database.query", 
                                 Encoding.UTF8.GetBytes("SELECT * FROM users"));

// Сервис-оркестратор заказов
async Task<OrderResult> ProcessOrder(Order order)
{
    // Шаг 1: Резервируем товары
    try
    {
        var inventoryRequest = new InventoryRequest { Items = order.Items };
        var inventoryResponse = await connection.RequestAsync<InventoryRequest, InventoryResponse>(
            "inventory.reserve", inventoryRequest);
        
        if (!inventoryResponse.Success)
            return new OrderResult { Success = false, Error = "Товары недоступны" };
        
        // Шаг 2: Обрабатываем платеж
        var paymentRequest = new PaymentRequest { 
            OrderId = order.Id, 
            Amount = order.TotalAmount 
        };
        var paymentResponse = await connection.RequestAsync<PaymentRequest, PaymentResponse>(
            "payment.process", paymentRequest);
        
        if (!paymentResponse.Success)
        {
            // Компенсирующая транзакция: отменяем резервацию
            await connection.PublishAsync("inventory.cancel-reservation", 
                new CancelReservationRequest { OrderId = order.Id });
            return new OrderResult { Success = false, Error = "Ошибка платежа" };
        }
        
        // Шаг 3: Создаем заявку на доставку
        var shippingRequest = new ShippingRequest { 
            OrderId = order.Id, 
            Address = order.ShippingAddress 
        };
        var shippingResponse = await connection.RequestAsync<ShippingRequest, ShippingResponse>(
            "shipping.schedule", shippingRequest);
        
        if (!shippingResponse.Success)
        {
            // Отменяем платеж и резервацию
            await connection.PublishAsync("payment.refund", 
                new RefundRequest { PaymentId = paymentResponse.PaymentId });
            await connection.PublishAsync("inventory.cancel-reservation", 
                new CancelReservationRequest { OrderId = order.Id });
            return new OrderResult { Success = false, Error = "Ошибка доставки" };
        }
        
        return new OrderResult { Success = true, OrderId = order.Id };
    }
    catch (Exception ex)
    {
        // Глобальная обработка ошибок и компенсация
        // ...
        return new OrderResult { Success = false, Error = ex.Message };
    }
}

// Создаем простой Circuit Breaker для NATS-запросов
public class NatsCircuitBreaker<TRequest, TResponse>
{
    private readonly INatsConnection _connection;
    private readonly string _subject;
    private readonly TimeSpan _timeout;
    private readonly int _maxFailures;
    private readonly TimeSpan _resetTimeout;
    
    private int _failureCount;
    private DateTime _lastFailure = DateTime.MinValue;
    private bool _isOpen;
 
    public NatsCircuitBreaker(INatsConnection connection, string subject, 
                            TimeSpan timeout, int maxFailures, TimeSpan resetTimeout)
    {
        _connection = connection;
        _subject = subject;
        _timeout = timeout;
        _maxFailures = maxFailures;
        _resetTimeout = resetTimeout;
    }
 
    public async Task<TResponse> ExecuteAsync(TRequest request)
    {
        // Проверяем, не открыт ли Circuit Breaker
        if (_isOpen)
        {
            // Проверяем, прошло ли достаточно времени для сброса
            if (DateTime.UtcNow - _lastFailure > _resetTimeout)
            {
                _isOpen = false;
                _failureCount = 0;
            }
            else
            {
                throw new CircuitBreakerOpenException($"Circuit breaker для {_subject} открыт");
            }
        }
 
        try
        {
            // Выполняем запрос с таймаутом
            var response = await _connection.RequestAsync<TRequest, TResponse>(
                _subject, request, _timeout);
            
            // Успешный запрос сбрасывает счетчик ошибок
            _failureCount = 0;
            return response;
        }
        catch (NATSTimeoutException)
        {
            _lastFailure = DateTime.UtcNow;
            _failureCount++;
            
            // Если превышен порог ошибок, открываем Circuit Breaker
            if (_failureCount >= _maxFailures)
            {
                _isOpen = true;
            }
            
            throw new ServiceUnavailableException(
                $"Сервис {_subject} не ответил в течение {_timeout.TotalSeconds} секунд");
        }
    }
}

var circuitBreaker = new NatsCircuitBreaker<InventoryRequest, InventoryResponse>(
    connection, 
    "inventory.check", 
    TimeSpan.FromSeconds(2),
    3,  // максимум 3 ошибки подряд
    TimeSpan.FromMinutes(1)  // время сброса - 1 минута
);
 
try
{
    var response = await circuitBreaker.ExecuteAsync(
        new InventoryRequest { ProductId = "ABC123" });
    // Обработка успешного ответа
}
catch (CircuitBreakerOpenException)
{
    // Обработка ситуации, когда Circuit Breaker открыт
    // Например, использование кеша или резервного сервиса
}
catch (ServiceUnavailableException)
{
    // Обработка единичного таймаута
}

using NATS.Client;
using NATS.Client.JetStream;
 
// Подключение к NATS
var cf = new ConnectionFactory();
var conn = cf.CreateConnection("nats://localhost:4222");
 
// Создание контекста JetStream
IJetStream js = conn.CreateJetStreamContext();

// Настройка потока
var streamConfig = StreamConfiguration.Builder()
    .WithName("orders")
    .WithSubjects("orders.*")
    .WithStorage(StorageType.File)
    .WithRetentionPolicy(RetentionPolicy.Limits)
    .WithMaxMessages(10000)
    .Build();
 
// Создание потока
StreamInfo streamInfo = js.AddStream(streamConfig);

// Публикация сообщения с подтверждением
var ack = js.Publish("orders.new", Encoding.UTF8.GetBytes(orderJson));
Console.WriteLine($"Сообщение сохранено с последовательным номером {ack.Seq} в потоке {ack.Stream}");

// Создание pull-потребителя
var consumerConfig = ConsumerConfiguration.Builder()
    .WithDurable("order-processor") // имя потребителя
    .WithFilterSubject("orders.new")
    .WithAckWait(Duration.OfSeconds(30))
    .Build();
 
js.AddOrUpdateConsumer("orders", consumerConfig);
 
// Получение сообщений через pull-модель
var batch = js.Fetch("orders", "order-processor", 10, Duration.OfSeconds(5));
foreach (var msg in batch)
{
    var order = JsonSerializer.Deserialize<Order>(Encoding.UTF8.GetString(msg.Data));
    ProcessOrder(order);
    msg.Ack(); // Явное подтверждение обработки
}

// Создание push-потребителя и подписка
var pushConsumerConfig = PushSubscribeOptions.Builder()
    .WithDurable("notifications-handler")
    .WithConfiguration(
        ConsumerConfiguration.Builder()
            .WithAckPolicy(AckPolicy.Explicit)
            .Build()
    ).Build();
 
// Подписка на сообщения через push-модель
js.PushSubscribeAsync("orders.completed", (sender, args) =>
{
    var order = JsonSerializer.Deserialize<Order>(
        Encoding.UTF8.GetString(args.Message.Data));
    SendNotification(order);
    args.Message.Ack();
}, false, pushConsumerConfig);

// Расширенный пример push-подписки с обработкой ошибок
var pushOpts = PushSubscribeOptions.Builder()
    .WithDurable("alerts-processor")
    .WithConfiguration(
        ConsumerConfiguration.Builder()
            .WithMaxDeliver(3)  // Максимальное количество попыток доставки
            .WithAckWait(Duration.OfSeconds(10))  // Время ожидания подтверждения
            .Build()
    ).Build();
 
var subscription = js.PushSubscribeAsync("system.alerts", (sender, args) =>
{
    try
    {
        var alert = JsonSerializer.Deserialize<SystemAlert>(
            Encoding.UTF8.GetString(args.Message.Data));
        
        ProcessAlert(alert);
        args.Message.Ack();  // Подтверждаем успешную обработку
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Ошибка обработки: {ex.Message}");
        args.Message.Nak();  // Отклоняем сообщение для повторной доставки
    }
}, false, pushOpts);

// Более сложный пример pull-подписки с повторными попытками
var consumer = ConsumerConfiguration.Builder()
    .WithDurable("batch-processor")
    .WithDeliverPolicy(DeliverPolicy.All)  // Получать все сообщения, включая архивные
    .WithFilterSubject("orders.processed")
    .WithMaxAckPending(100)  // Лимит на количество неподтвержденных сообщений
    .Build();
 
js.AddOrUpdateConsumer("orders", consumer);
 
// Цикл обработки сообщений с повторными попытками
while (isRunning)
{
    try
    {
        var messages = js.Fetch("orders", "batch-processor", 20, Duration.OfSeconds(3));
        
        foreach (var msg in messages)
        {
            try
            {
                var order = JsonSerializer.Deserialize<Order>(
                    Encoding.UTF8.GetString(msg.Data));
                
                await ProcessOrderAsync(order);
                msg.Ack();
            }
            catch (TemporaryException ex)
            {
                // Для временных ошибок используем Nak() для повторной доставки
                Console.WriteLine($"Временная ошибка: {ex.Message}");
                msg.Nak();
            }
            catch (Exception ex)
            {
                // Для критических ошибок используем Term() - больше не пытаться доставить
                Console.WriteLine($"Критическая ошибка: {ex.Message}");
                msg.Term();
            }
        }
        
        // Делаем паузу, если обработали меньше сообщений, чем запрашивали
        if (messages.Count < 20)
            await Task.Delay(1000);
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Ошибка получения сообщений: {ex.Message}");
        await Task.Delay(5000);  // Пауза перед следующей попыткой
    }
}

var streamConfig = StreamConfiguration.Builder()
    .WithName("logs")
    .WithSubjects("logs.*")
    .WithRetentionPolicy(RetentionPolicy.Limits)
    .WithMaxAge(Duration.OfDays(7))     // Хранить не более 7 дней
    .WithMaxBytes(1024 * 1024 * 1024)   // Ограничение размером в 1 ГБ
    .WithMaxMessages(1_000_000)         // Не более миллиона сообщений
    .WithNoWrap(false)                  // Разрешить перезапись старых сообщений
    .Build();

var streamConfig = StreamConfiguration.Builder()
    .WithName("user-profiles")
    .WithSubjects("user.*.profile")
    .WithRetentionPolicy(RetentionPolicy.Limits)
    .WithMaxMessages(-1)        // Нет ограничения по количеству
    .WithMaxBytes(-1)           // Нет ограничения по размеру
    .WithMaxAge(Duration.OfDays(365))  // Хранить год
    .WithAllowRollup(true)      // Включаем возможность компактификации
    .Build();

var headers = new MsgHeader();
headers.Add("Nats-Rollup", "sub");  // Заменяет все предыдущие сообщения с этим субъектом
 
var userData = JsonSerializer.Serialize(updatedProfile);
js.Publish("user.12345.profile", headers, Encoding.UTF8.GetBytes(userData));

public void ConfigureServices(IServiceCollection services)
{
    // Другие сервисы...
    
    // Регистрация NATS-клиента как Singleton
    services.AddSingleton<INatsConnection>(provider =>
    {
        var configuration = provider.GetRequiredService<IConfiguration>();
        var natsUrl = configuration["Nats:Url"] ?? "nats://localhost:4222";
        
        var opts = ConnectionFactory.GetDefaultOptions();
        opts.Url = natsUrl;
        opts.Name = "asp-net-core-service";
        
        var connection = new ConnectionFactory().CreateConnection(opts);
        return connection;
    });
    
    // Регистрация обертки над JetStream (опционально)
    services.AddSingleton<INatsJetStreamClient>(provider =>
    {
        var connection = provider.GetRequiredService<INatsConnection>();
        return new NatsJetStreamClient(connection);
    });
    
    // Регистрация наших NATS-зависимых сервисов
    services.AddScoped<INotificationService, NatsNotificationService>();
}

public class NatsNotificationService : INotificationService
{
    private readonly INatsConnection _connection;
    
    public NatsNotificationService(INatsConnection connection)
    {
        _connection = connection;
    }
    
    public async Task SendNotificationAsync(string userId, string message)
    {
        var notification = new Notification
        {
            UserId = userId,
            Message = message,
            Timestamp = DateTime.UtcNow
        };
        
        string json = JsonSerializer.Serialize(notification);
        await _connection.PublishAsync($"notification.{userId}", 
                                     Encoding.UTF8.GetBytes(json));
    }
}

public class NatsHostedService : IHostedService
{
    private readonly INatsConnection _connection;
    private readonly ILogger<NatsHostedService> _logger;
    private List<IAsyncSubscription> _subscriptions = new();
    
    public NatsHostedService(INatsConnection connection, ILogger<NatsHostedService> logger)
    {
        _connection = connection;
        _logger = logger;
    }
    
    public Task StartAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Запуск NATS сервиса");
        
        // Настраиваем подписки
        var subscription = _connection.SubscribeAsync("service.commands", HandleCommand);
        _subscriptions.Add(subscription);
        
        return Task.CompletedTask;
    }
    
    public Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Остановка NATS сервиса");
        
        // Отписываемся от всех тем
        foreach (var sub in _subscriptions)
        {
            try
            {
                sub.Dispose();
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Ошибка при отписке от NATS");
            }
        }
        
        return Task.CompletedTask;
    }
    
    private void HandleCommand(object sender, MsgHandlerEventArgs e)
    {
        var command = Encoding.UTF8.GetString(e.Message.Data);
        _logger.LogInformation("Получена команда: {Command}", command);
        
        // Обработка команды...
    }
}

services.AddHostedService<NatsHostedService>();

// Отправка пакета сообщений
var messages = new List<OrderEvent>(100);
// Заполняем список сообщениями...
 
var batch = connection.CreateBatch();
foreach (var msg in messages)
{
    string json = JsonSerializer.Serialize(msg);
    batch.Add("orders.events", Encoding.UTF8.GetBytes(json));
}
batch.Publish();

// Использование MessagePack вместо JSON
using MessagePack;
 
// Настройка сериализатора
var options = MessagePackSerializerOptions.Standard;
 
// Сериализация и публикация
var data = MessagePackSerializer.Serialize(order, options);
connection.Publish("orders.created", data);

// Использование ArrayPool для повторного использования буферов
byte[] buffer = ArrayPool<byte>.Shared.Rent(maxMessageSize);
try
{
    int bytesWritten = SerializeMessage(message, buffer);
    connection.Publish("data.large", buffer.AsSpan(0, bytesWritten).ToArray());
}
finally
{
    ArrayPool<byte>.Shared.Return(buffer);
}

// Неоптимальный обработчик - создаёт много объектов
connection.Subscribe("data.process", (sender, args) =>
{
    var data = Encoding.UTF8.GetString(args.Message.Data);  // Создаёт новую строку
    var items = data.Split(',');  // Создаёт новый массив
    var values = items.Select(i => int.Parse(i)).ToList();  // Создаёт список и боксинг
    // Обработка...
});
 
// Оптимизированный обработчик
connection.Subscribe("data.process", (sender, args) =>
{
    // Обработка напрямую, без создания промежуточных строк
    ReadOnlySpan<byte> dataSpan = args.Message.Data;
    // Обработка без лишних аллокаций...
});

try
{
    await connection.PublishAsync("critical.data", messageData);
}
catch (NATSConnectionException ex)
{
    // Проблемы с соединением - может иметь смысл попробовать переподключиться
    _logger.LogError(ex, "Потеряно соединение с NATS");
    await ReconnectAsync();
    // Повторная попытка или запись в локальную очередь
}
catch (NATSTimeoutException ex)
{
    // Таймаут - возможно, сервер перегружен
    _logger.LogWarning(ex, "Таймаут при публикации в NATS");
    // Экспоненциальная задержка перед повторной попыткой
    await Task.Delay(CalculateBackoff(retryAttempt));
}
catch (Exception ex)
{
    // Прочие неожиданные ошибки
    _logger.LogCritical(ex, "Критическая ошибка при работе с NATS");
    // Уведомление оператора, запись в dead-letter queue и т.д.
}

// Пример использования Polly для реализации Circuit Breaker
var circuitBreaker = Policy
    .Handle<NATSException>()
    .CircuitBreakerAsync(
        exceptionsAllowedBeforeBreaking: 3,
        durationOfBreak: TimeSpan.FromMinutes(1)
    );
 
await circuitBreaker.ExecuteAsync(async () =>
{
    await connection.PublishAsync("critical.operation", data);
});

// В RabbitMQ: публикация в обменник с routing key
channel.BasicPublish(
    exchange: "orders",
    routingKey: "order.created.highpriority",
    body: Encoding.UTF8.GetBytes(orderJson)
);
 
// Эквивалент в NATS: использование иерархического субъекта
natsConnection.Publish(
    "orders.created.highpriority",
    Encoding.UTF8.GetBytes(orderJson)
);

// В Kafka: чтение с указанием смещения
var consumeResult = consumer.Consume(cancellationToken);
// Обработка сообщения...
consumer.Commit(consumeResult);
 
// В NATS JetStream: настройка потребителя по времени
var consumerConfig = ConsumerConfiguration.Builder()
    .WithDeliverPolicy(DeliverPolicy.ByStartTime)
    .WithStartTime(DateTime.UtcNow.AddDays(-1))
    .Build();

// Пример мостового сервиса
public class MessageBridge
{
    private readonly IKafkaConsumer _kafkaConsumer;
    private readonly INatsConnection _natsConnection;
 
    public async Task BridgeMessages()
    {
        var message = await _kafkaConsumer.ConsumeAsync();
        // Копируем сообщение в NATS с сохранением ключа партиции в субъекте
        await _natsConnection.PublishAsync(
            $"bridge.{message.Topic}.{message.Partition}",
            message.Value
        );
    }
}

// NatsServiceExtensions.cs
public static class NatsServiceExtensions
{
    public static IServiceCollection AddNatsServices(
        this IServiceCollection services, 
        IConfiguration configuration)
    {
        var natsUrl = configuration["Nats:Url"] ?? "nats://localhost:4222";
        
        services.AddSingleton<INatsConnection>(sp =>
        {
            var opts = ConnectionFactory.GetDefaultOptions();
            opts.Url = natsUrl;
            opts.Name = sp.GetRequiredService<IHostEnvironment>().ApplicationName;
            opts.AllowReconnect = true;
            opts.MaxReconnect = 5;
            
            var logger = sp.GetRequiredService<ILogger<INatsConnection>>();
            opts.DisconnectedEventHandler = (sender, args) => 
                logger.LogWarning("Соединение с NATS потеряно: {Reason}", args.Reason);
            opts.ReconnectedEventHandler = (sender, args) => 
                logger.LogInformation("Переподключено к NATS: {Url}", args.ConnectedUrl);
            
            return new ConnectionFactory().CreateConnection(opts);
        });
        
        services.AddSingleton<IJetStreamContext>(sp =>
        {
            var conn = sp.GetRequiredService<INatsConnection>();
            return conn.CreateJetStreamContext();
        });
        
        return services;
    }
}

// OrderService.cs
public class OrderService : IOrderService
{
    private readonly INatsConnection _nats;
    private readonly IJetStreamContext _js;
    private readonly ILogger<OrderService> _logger;
    
    public OrderService(
        INatsConnection nats, 
        IJetStreamContext js,
        ILogger<OrderService> logger)
    {
        _nats = nats;
        _js = js;
        _logger = logger;
        
        // Создаем поток для хранения событий заказов
        try
        {
            _js.AddStream(StreamConfiguration.Builder()
                .WithName("ORDERS")
                .WithSubjects("orders.*")
                .WithStorage(StorageType.File)
                .Build());
        }
        catch (NATSJetStreamException ex) when (ex.ErrorCode == 10058) // Stream already exists
        {
            // Поток уже существует, игнорируем
        }
    }
    
    public async Task<OrderResult> CreateOrderAsync(OrderRequest request)
    {
        // Валидация запроса, бизнес-логика и т.д.
        
        // Создаем заказ в базе данных
        var order = new Order
        {
            Id = Guid.NewGuid().ToString("N"),
            CustomerId = request.CustomerId,
            Items = request.Items,
            TotalAmount = request.Items.Sum(i => i.Price * i.Quantity),
            Status = OrderStatus.Created,
            CreatedAt = DateTime.UtcNow
        };
        
        // Сохраняем в БД (код опущен для краткости)
        
        // Публикуем событие создания заказа
        var orderCreatedEvent = new OrderCreatedEvent
        {
            OrderId = order.Id,
            CustomerId = order.CustomerId,
            TotalAmount = order.TotalAmount,
            Items = order.Items.Select(i => new OrderItemDto
            {
                ProductId = i.ProductId,
                Quantity = i.Quantity,
                Price = i.Price
            }).ToList()
        };
        
        await _js.PublishAsync(
            "orders.created",
            JsonSerializer.SerializeToUtf8Bytes(orderCreatedEvent)
        );
        
        _logger.LogInformation("Создан заказ {OrderId} на сумму {Amount}", 
            order.Id, order.TotalAmount);
            
        return new OrderResult
        {
            Success = true,
            OrderId = order.Id
        };
    }
}

// PaymentProcessor.cs
public class PaymentProcessor : BackgroundService
{
    private readonly IJetStreamContext _js;
    private readonly IPaymentGateway _paymentGateway;
    private readonly ILogger<PaymentProcessor> _logger;
    
    public PaymentProcessor(
        IJetStreamContext js,
        IPaymentGateway paymentGateway,
        ILogger<PaymentProcessor> logger)
    {
        _js = js;
        _paymentGateway = paymentGateway;
        _logger = logger;
    }
    
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Настраиваем потребителя с явным подтверждением
        var consumerOpts = ConsumerConfiguration.Builder()
            .WithDurable("payment-processor")
            .WithAckWait(Duration.OfSeconds(30))
            .WithAckPolicy(AckPolicy.Explicit)
            .Build();
            
        _js.AddOrUpdateConsumer("ORDERS", consumerOpts);
        
        // Подписываемся на события создания заказов
        var subscription = await _js.PushSubscribeAsync(
            "orders.created",
            "payment-processor",
            stoppingToken: stoppingToken);
            
        await foreach (var msg in subscription.Messages)
        {
            try
            {
                var orderEvent = JsonSerializer.Deserialize<OrderCreatedEvent>(
                    msg.Data);
                    
                // Обрабатываем платеж
                var paymentResult = await _paymentGateway.ProcessPaymentAsync(
                    orderEvent.CustomerId,
                    orderEvent.TotalAmount,
                    $"Оплата заказа {orderEvent.OrderId}");
                    
                if (paymentResult.Success)
                {
                    // Публикуем событие успешной оплаты
                    await _js.PublishAsync(
                        $"orders.payment.succeeded",
                        JsonSerializer.SerializeToUtf8Bytes(new PaymentSucceededEvent
                        {
                            OrderId = orderEvent.OrderId,
                            PaymentId = paymentResult.PaymentId,
                            Amount = orderEvent.TotalAmount,
                            ProcessedAt = DateTime.UtcNow
                        })
                    );
                    
                    _logger.LogInformation(
                        "Оплачен заказ {OrderId}, платеж {PaymentId}",
                        orderEvent.OrderId, paymentResult.PaymentId);
                }
                else
                {
                    // Публикуем событие неудачной оплаты
                    await _js.PublishAsync(
                        $"orders.payment.failed",
                        JsonSerializer.SerializeToUtf8Bytes(new PaymentFailedEvent
                        {
                            OrderId = orderEvent.OrderId,
                            Reason = paymentResult.ErrorMessage,
                            FailedAt = DateTime.UtcNow
                        })
                    );
                    
                    _logger.LogWarning(
                        "Не удалось оплатить заказ {OrderId}: {Reason}",
                        orderEvent.OrderId, paymentResult.ErrorMessage);
                }
                
                // Подтверждаем обработку сообщения
                await msg.AckAsync();
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Ошибка обработки платежа");
                // Отклоняем сообщение для повторной обработки
                await msg.NakAsync();
            }
        }
    }
}

// NotificationService.cs
public class NotificationService : BackgroundService
{
    private readonly INatsConnection _nats;
    private readonly IEmailSender _emailSender;
    private readonly ILogger<NotificationService> _logger;
    private List<IAsyncSubscription> _subscriptions = new();
 
    public NotificationService(
        INatsConnection nats,
        IEmailSender emailSender,
        ILogger<NotificationService> logger)
    {
        _nats = nats;
        _emailSender = emailSender;
        _logger = logger;
    }
 
    protected override Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Подписываемся на события успешной оплаты
        var paymentSucceededSub = _nats.SubscribeAsync(
            "orders.payment.succeeded", 
            async (sender, args) =>
            {
                var evt = JsonSerializer.Deserialize<PaymentSucceededEvent>(
                    args.Message.Data);
                
                await _emailSender.SendOrderConfirmationAsync(
                    evt.OrderId, 
                    evt.Amount);
                
                _logger.LogInformation(
                    "Отправлено подтверждение заказа {OrderId}", 
                    evt.OrderId);
            });
        
        _subscriptions.Add(paymentSucceededSub);
        
        // Подписка на события неудачной оплаты
        var paymentFailedSub = _nats.SubscribeAsync(
            "orders.payment.failed", 
            async (sender, args) =>
            {
                var evt = JsonSerializer.Deserialize<PaymentFailedEvent>(
                    args.Message.Data);
                
                await _emailSender.SendPaymentFailedNotificationAsync(
                    evt.OrderId, 
                    evt.Reason);
                
                _logger.LogInformation(
                    "Отправлено уведомление о неудачной оплате {OrderId}", 
                    evt.OrderId);
            });
        
        _subscriptions.Add(paymentFailedSub);
        
        return Task.CompletedTask;
    }
 
    public override Task StopAsync(CancellationToken cancellationToken)
    {
        foreach (var sub in _subscriptions)
        {
            try { sub.Dispose(); }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Ошибка при отписке");
            }
        }
        
        return base.StopAsync(cancellationToken);
    }
}

// OrdersController.cs в API Gateway
[ApiController]
[Route("api/[controller]")]
public class OrdersController : ControllerBase
{
    private readonly INatsConnection _nats;
    private readonly ILogger<OrdersController> _logger;
 
    public OrdersController(
        INatsConnection nats,
        ILogger<OrdersController> logger)
    {
        _nats = nats;
        _logger = logger;
    }
 
    [HttpPost]
    public async Task<IActionResult> CreateOrder(OrderRequest request)
    {
        if (!ModelState.IsValid)
            return BadRequest(ModelState);
 
        try
        {
            // Используем Request-Reply паттерн для взаимодействия с сервисом заказов
            var response = await _nats.RequestAsync<OrderRequest, OrderResult>(
                "order.create", 
                request, 
                timeout: TimeSpan.FromSeconds(10));
            
            if (response.Success)
                return Ok(new { OrderId = response.OrderId });
            else
                return BadRequest(new { Error = response.Error });
        }
        catch (NATSTimeoutException)
        {
            _logger.LogError("Таймаут при обработке заказа");
            return StatusCode(503, new { Error = "Сервис временно недоступен" });
        }
    }
}