4. Request Reply паттерн

    Почтовая аналогия Request-Reply:
    Отправляя письмо с вопросом в компанию, вы вкладываете внутрь
    конверт с вашим обратным адресом и наклеенной маркой (Reply Subject).
    Получатель читает вопрос, пишет ответ, запечатывает его в ваш
    обратный конверт и опускает в ближайший почтовый ящик.

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "time"

    "github.com/nats-io/nats.go"
)

type User struct {
    ID      string `json:"id"`
    Name    string `json:"name"`
    Email   string `json:"email"`
    Created string `json:"created_at"`
}

func main() {
    nc, err := nats.Connect(nats.DefaultURL)
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Close()

    // 1. Запускаем обработчик запросов (сервер)
    go startUserServer(nc)

    // Даем серверу миллисекунду на регистрацию подписки
    time.Sleep(50 * time.Millisecond)

    // 2. Выполняем синхронные запросы (клиент)
    makeRequests(nc)
}

func startUserServer(nc *nats.Conn) {
    // Подписываемся на субъект запросов
    _, err := nc.Subscribe("api.get.user", func(msg *nats.Msg) {
        userID := string(msg.Data)

        // Формируем полезную нагрузку ответа
        user := User{
            ID:      userID,
            Name:    "John Doe",
            Email:   "john@example.com",
            Created: time.Now().Format(time.RFC3339),
        }

        response, err := json.Marshal(user)
        if err != nil {
            log.Printf("Marshal error: %v", err)
            return
        }

        // Отправляем ответ на временный reply subject клиента
        if msg.Reply != "" {
            err = nc.Publish(msg.Reply, response)
            if err != nil {
                log.Printf("Failed to send reply: %v", err)
            }
        }
    })
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("User server responder started on [api.get.user]...")
    select {} // Удерживаем фоновую горутину
}

func makeRequests(nc *nats.Conn) {
    for i := 1; i <= 3; i++ {
        userID := fmt.Sprintf("user-%d", i)

        // nc.Request блокирует горутину до получения ответа или истечения таймаута
        resp, err := nc.Request("api.get.user", []byte(userID), 2*time.Second)
        if err != nil {
            log.Printf("Request failed: %v", err)
            continue
        }

        var user User
        if err := json.Unmarshal(resp.Data, &user); err != nil {
            log.Printf("Failed to parse response: %v", err)
            continue
        }

        fmt.Printf("Received user: %+v\n", user)
    }
}

func robustRequestExample(nc *nats.Conn) {
    userID := "test-user"

    // Устанавливаем жесткий таймаут в 5 секунд
    resp, err := nc.Request("api.get.user", []byte(userID), 5*time.Second)
    if err != nil {
        switch err {
        case nats.ErrTimeout:
            log.Printf("Request timeout after 5 seconds: сервис не ответил вовремя")
        case nats.ErrNoResponders:
            log.Printf("No service available to handle request: нет активных обработчиков в кластере!")
        default:
            log.Printf("Request failed: %v", err)
        }
        return
    }

    fmt.Printf("Success response: %s\n", string(resp.Data))
}

func concurrentRequests(nc *nats.Conn) {
    const numRequests = 10
    responses := make(chan *nats.Msg, numRequests)
    errors := make(chan error, numRequests)

    // Запускаем пул параллельных горутин
    for i := 0; i < numRequests; i++ {
        go func(requestID int) {
            resp, err := nc.Request(
                "api.get.user",
                []byte(fmt.Sprintf("user-%d", requestID)),
                3*time.Second,
            )

            if err != nil {
                errors <- err
                return
            }

            responses <- resp
        }(i)
    }

    // Асинхронно собираем результаты
    for i := 0; i < numRequests; i++ {
        select {
        case resp := <-responses:
            fmt.Printf("Got response: %s\n", string(resp.Data))
        case err := <-errors:
            fmt.Printf("Got error: %v\n", err)
        }
    }
}

// HTTP Request: требует резолва DNS, создания/извлечения соединения из пула, формирования HTTP-заголовков
resp, err := http.Get("http://user-service/api/user/123")

// NATS Request-Reply: запись пакета в уже открытый TCP-сокет по имени логического субъекта
resp, err := nc.Request("api.get.user", []byte("123"), 2*time.Second)

// На каждом инстансе сервиса запускается код:
nc.QueueSubscribe("api.get.user", "user-service-workers", func(msg *nats.Msg) {
    // Обработка ровно одним экземпляром из группы
    nc.Publish(msg.Reply, []byte(`{"status": "ok"}`))
})

func createAuthenticatedRequestHandler(nc *nats.Conn, next nats.MsgHandler) nats.MsgHandler {
    return func(msg *nats.Msg) {
        // 1. Проверяем токен авторизации из заголовка
        token := msg.Header.Get("Authorization")
        if token != "secret-token-123" {
            if msg.Reply != "" {
                _ = nc.Publish(msg.Reply, []byte(`{"error": "unauthorized"}`))
            }
            return
        }

        // 2. Замеряем время выполнения
        start := time.Now()

        // 3. Передаем управление бизнес-логике
        next(msg)

        // 4. Логируем результат
        duration := time.Since(start)
        log.Printf("[METRIC] Subject: %s handled in %v", msg.Subject, duration)
    }
}

// Использование middleware
userHandler := func(msg *nats.Msg) {
    _ = nc.Publish(msg.Reply, []byte(`{"user": "Alice"}`))
}

_, err = nc.Subscribe("api.get.user", createAuthenticatedRequestHandler(nc, userHandler))

import (
    "context"
    "time"
)

func requestWithContext(ctx context.Context, nc *nats.Conn, subject string, data []byte) (*nats.Msg, error) {
    // Если клиентский контекст уже содержит дедлайн или отмену,
    // RequestWithContext корректно прервет ожидание без утечки горутин
    return nc.RequestWithContext(ctx, subject, data)
}

func legacyRequestWithContext(ctx context.Context, nc *nats.Conn, subject string, data []byte) (*nats.Msg, error) {
    result := make(chan struct {
        msg *nats.Msg
        err error
    }, 1)

    go func() {
        msg, err := nc.Request(subject, data, 30*time.Second)
        result <- struct {
            msg *nats.Msg
            err error
        }{msg, err}
    }()

    select {
    case res := <-result:
        return res.msg, res.err
    case <-ctx.Done():
        return nil, ctx.Err()
    }
}

func benchmarkRequestReply() {
    nc, _ := nats.Connect(nats.DefaultURL)

    // Прогрев (Warm up)
    for i := 0; i < 100; i++ {
        _, _ = nc.Request("api.ping", []byte("ping"), time.Second)
    }

    // Замер производительности
    start := time.Now()
    const requests = 10000

    for i := 0; i < requests; i++ {
        _, err := nc.Request("api.ping", []byte("ping"), time.Second)
        if err != nil {
            log.Fatal(err)
        }
    }

    duration := time.Since(start)
    avgLatency := duration.Nanoseconds() / int64(requests)

    fmt.Printf("Total: %v, Avg latency: %dns per request\n", duration, avgLatency)
    // В локальной сети / localhost: ~50-90µs на запрос (против ~1-5ms для чистого HTTP REST)
}

// Хорошо: Семантически понятные и масштабируемые subject'ы
"service.users.api.get"
"service.orders.api.create"

// Плохо: Жестко закодированные или бессистемные имена
"get_user_by_id_123"
"service_abc_def_ghi"