Создание микросервисов с gRPC и Protobuf в C++

// Простой вызов gRPC метода выглядит почти как обычная функция
UserResponse response = stub->GetUser(&context, request);

// Пример ошибки, которая будет поймана компилятором в gRPC
// но может проявиться только в рантайме при использовании REST+JSON
request.set_user_id(42); // Ошибка: user_id имеет тип string, а не int

ClientContext context;
std::chrono::system_clock::time_point deadline = 
    std::chrono::system_clock::now() + std::chrono::seconds(5);
context.set_deadline(deadline);
 
// Если метод не выполнится за 5 секунд, запрос будет отменён
UserResponse response = stub->GetUser(&context, request);

// Глобальное сжатие для всех вызовов через этот канал
ChannelArguments args;
args.SetCompressionAlgorithm(GRPC_COMPRESS_GZIP);
auto channel = CreateCustomChannel(
    server_address, InsecureChannelCredentials(), args);
 
// Или для конкретного вызова
ClientContext context;
context.set_compression_algorithm(GRPC_COMPRESS_GZIP);

// Пример настройки канала с клиентской балансировкой в C++
auto channel = CreateChannel("my-service", LoadBalancingPolicy());
auto stub = MyService::NewStub(channel);

Status StreamData(ServerContext* context, 
                 const Request* request,
                 ServerWriter<Response>* writer) {
    Response response;
    for (int i = 0; i < 10; i++) {
        response.set_value(i);
        writer->Write(response);
    }
    return Status::OK;
}

ClientContext context;
Request request;
Response response;
std::unique_ptr<ClientReader<Response>> reader(
    stub->StreamData(&context, request));
    
while (reader->Read(&response)) {
    // Обработка каждого ответа по мере поступления
    std::cout << "Получено: " << response.value() << std::endl;
}
 
Status status = reader->Finish();

ClientContext context;
std::chrono::system_clock::time_point deadline =
    std::chrono::system_clock::now() + std::chrono::milliseconds(500);
context.set_deadline(deadline);
 
// Сервер получит уведомление о дедлайне и сможет  
// соответствующим образом адаптировать свою работу
UserResponse response = stub->GetUser(&context, request);

Status GetUser(ServerContext* context, const UserRequest* request,
               UserResponse* response) {
    if (context->IsCancelled()) {
        return Status(StatusCode::CANCELLED, "Запрос отменён клиентом");
    }
    
    // Пример проверки, сколько времени осталось до дедлайна
    if (context->deadline() < std::chrono::system_clock::now()) {
        return Status(StatusCode::DEADLINE_EXCEEDED, "Превышен дедлайн");
    }
    
    // Обработка запроса...
    return Status::OK;
}

// Пример интеграции с OpenCensus для трассировки
auto tracer = ::opencensus::trace::Tracer::GetHandle(
    "my.namespace/MyTracer");
auto span = tracer->StartSpan("MyOperation");
{
    opencensus::trace::WithSpan ws(span);
    // Код, выполняемый в контексте этого span
    // gRPC автоматически подхватит контекст трассировки
    UserResponse response = stub->GetUser(&context, request);
}
span->End();

syntax = "proto3"; // Указываем версию синтаксиса
 
package users.management; // Определяем пакет для предотвращения конфликтов имен
 
// Импорт определений из других .proto файлов
import "common/types.proto";
 
// Определяем сервис - набор методов, которые можно вызывать удаленно
service UserManagement {
  // Унарный метод: один запрос, один ответ
  rpc GetUser(GetUserRequest) returns (User);
  
  // Серверный потоковый метод: один запрос, поток ответов
  rpc ListUsers(ListUsersRequest) returns (stream User);
  
  // Клиентский потоковый метод: поток запросов, один ответ
  rpc BatchCreateUsers(stream CreateUserRequest) returns (BatchResponse);
  
  // Двунаправленный потоковый метод: оба участника отправляют потоки сообщений
  rpc ChatWithSupport(stream ChatMessage) returns (stream ChatMessage);
}
 
// Определение сообщения - структуры данных
message User {
  string id = 1; // Каждое поле имеет уникальный номер (тэг)
  string name = 2;
  string email = 3;
  UserStatus status = 4; // Использование перечисления
  repeated string roles = 5; // Массив строк
  map<string, string> metadata = 6; // Ассоциативный массив
  
  // Вложенный тип, видимый только внутри User
  message Address {
    string street = 1;
    string city = 2;
    string postal_code = 3;
    string country = 4;
  }
  
  repeated Address addresses = 7; // Массив вложенных объектов
  
  oneof contact { // Только одно из полей может быть установлено
    string phone_number = 8;
    string alternative_email = 9;
  }
  
  common.Timestamp created_at = 10; // Импортированный тип
}
 
// Перечисление - набор именованных констант
enum UserStatus {
  UNKNOWN = 0; // Первое значение должно быть 0
  ACTIVE = 1;
  SUSPENDED = 2;
  DELETED = 3;
}
 
// Другие сообщения для запросов и ответов
message GetUserRequest {
  string user_id = 1;
}
 
message ListUsersRequest {
  int32 page_size = 1;
  string page_token = 2;
  string filter = 3;
}
 
message CreateUserRequest {
  User user = 1;
}
 
message BatchResponse {
  int32 success_count = 1;
  int32 failure_count = 2;
  repeated string error_messages = 3;
}
 
message ChatMessage {
  string sender = 1;
  string content = 2;
  common.Timestamp sent_at = 3;
}

// Создание объекта и установка полей
users::management::User user;
user.set_id("user123");
user.set_name("Иван Петров");
user.set_email("ivan@example.com");
user.set_status(users::management::UserStatus::ACTIVE);
user.add_roles("admin"); // Добавление элемента в repeated поле
user.add_roles("editor");
(*user.mutable_metadata())["department"] = "Engineering"; // Работа с map
 
// Добавление адреса (вложенный объект)
auto* address = user.add_addresses();
address->set_street("ул. Пушкина, 10");
address->set_city("Москва");
address->set_postal_code("123456");
address->set_country("Россия");
 
// Работа с oneof - установка одного из взаимоисключающих полей
user.set_phone_number("+7 123 456 7890");
// Теперь phone_number установлен, а alternative_email - нет
// user.has_phone_number() вернет true
// user.has_alternative_email() вернет false
 
// Сериализация в строку
std::string serialized;
user.SerializeToString(&serialized);
 
// Десериализация
users::management::User user2;
user2.ParseFromString(serialized);

message User {
  // Все существующие поля сохраняются
  string id = 1;
  string name = 2;
  // ...
  
  // Новое поле
  bool is_verified = 11;
}

// Пользователь системы
message User {
  // Уникальный идентификатор пользователя
  // Должен соответствовать формату UUID v4
  string id = 1 [(validate.rules).string.pattern = "^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"];
  
  // Полное имя пользователя
  string name = 2 [(validate.rules).string.min_len = 2, (validate.rules).string.max_len = 100];
  
  // Email пользователя для связи
  string email = 3 [(validate.rules).string.email = true];
  
  // ...
}

// Zero-Copy десериализация
const void* data = // указатель на сериализованные данные
int size = // размер данных
google::protobuf::Arena arena;
users::management::User* user = 
    google::protobuf::Arena::CreateMessage<users::management::User>(&arena);
user->ParseFromArray(data, size);
 
// Все выделения памяти внутри user будут происходить 
// через arena и освобождаться все вместе при ее уничтожении

// Пример замера производительности сериализации
 
#include <benchmark/benchmark.h>
#include <json/json.h>
#include "user.pb.h"
 
static void ProtobufSerialize(benchmark::State& state) {
  users::management::User user;
  user.set_id("user123");
  user.set_name("Иван Петров");
  user.set_email("ivan@example.com");
  // ... заполняем другие поля ...
  
  for (auto _ : state) {
    std::string output;
    user.SerializeToString(&output);
    benchmark::DoNotOptimize(output);
  }
}
 
static void JsonSerialize(benchmark::State& state) {
  Json::Value user;
  user["id"] = "user123";
  user["name"] = "Иван Петров";
  user["email"] = "ivan@example.com";
  // ... заполняем другие поля ...
  
  Json::FastWriter writer;
  for (auto _ : state) {
    std::string output = writer.write(user);
    benchmark::DoNotOptimize(output);
  }
}
 
BENCHMARK(ProtobufSerialize);
BENCHMARK(JsonSerialize);

proto/
  ├── common/
  │     ├── types.proto
  │     ├── errors.proto
  │     └── pagination.proto
  ├── users/
  │     ├── user.proto
  │     └── profile.proto
  ├── products/
  │     ├── product.proto
  │     └── inventory.proto
  └── api/
        ├── users_service.proto
        └── products_service.proto

// Пример эволюции схемы с сохранением обратной совместимости
message User {
  // Существующие поля
  string id = 1;
  string name = 2;
  
  // Устаревшее поле - не используйте его
  string email = 3 [deprecated = true];
  
  // Новое поле, заменяющее устаревшее
  string contact_email = 4;
  
  // Полностью новое поле
  bool is_verified = 5;
}

// Потоковый парсинг большого сообщения
std::ifstream file("large_data.pb", 
                   std::ios::in | std::ios::binary);
 
google::protobuf::io::IstreamInputStream input(&file);
google::protobuf::io::CodedInputStream coded_input(&input);
 
// Устанавливаем лимиты для защиты от чрезмерного потребления памяти
coded_input.SetTotalBytesLimit(1024 * 1024 * 1024);  // 1 GB
 
users::management::User user;
bool success = user.ParseFromCodedStream(&coded_input);

// Использование Arena для эффективного управления памятью
google::protobuf::Arena arena;
 
// Создаем сообщение в арене
auto* user = google::protobuf::Arena::CreateMessage<users::management::User>(&arena);
user->set_id("user123");
user->set_name("Иван Петров");
 
// Все динамические выделения памяти внутри user 
// будут происходить через арену
auto* address = user->add_addresses();
address->set_city("Москва");
 
// Не нужно заботиться об освобождении памяти - 
// все будет освобождено при уничтожении арены

import "google/protobuf/wrappers.proto";
 
message User {
  string id = 1;
  google.protobuf.StringValue nickname = 2;  // Может быть явно null
}

my_microservice/
├── CMakeLists.txt
├── protos/
│   └── service.proto
├── src/
│   ├── server/
│   │   └── server.cc
│   └── client/
│       └── client.cc
└── include/
    └── common/
        └── utils.h

cmake_minimum_required(VERSION 3.13)
project(MyMicroservice CXX)
 
set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
# Найти gRPC и Protobuf
find_package(gRPC CONFIG REQUIRED)
find_package(Protobuf CONFIG REQUIRED)
 
# Генерация кода из .proto файлов
get_filename_component(proto_path "${CMAKE_SOURCE_DIR}/protos/service.proto" ABSOLUTE)
get_filename_component(proto_dir "${proto_path}" DIRECTORY)
 
set(generated_dir "${CMAKE_BINARY_DIR}/generated")
file(MAKE_DIRECTORY "${generated_dir}")
 
set(proto_srcs "${generated_dir}/service.pb.cc")
set(proto_hdrs "${generated_dir}/service.pb.h")
set(grpc_srcs "${generated_dir}/service.grpc.pb.cc")
set(grpc_hdrs "${generated_dir}/service.grpc.pb.h")
 
add_custom_command(
    OUTPUT "${proto_srcs}" "${proto_hdrs}" "${grpc_srcs}" "${grpc_hdrs}"
    COMMAND protobuf::protoc
    ARGS --grpc_out "${generated_dir}"
         --cpp_out "${generated_dir}"
         -I "${proto_dir}"
         --plugin=protoc-gen-grpc="$<TARGET_FILE:gRPC::grpc_cpp_plugin>"
         "${proto_path}"
    DEPENDS "${proto_path}")
 
# Серверное приложение
add_executable(server 
    src/server/server.cc 
    ${proto_srcs} 
    ${grpc_srcs})
target_include_directories(server PRIVATE "${generated_dir}" "${CMAKE_SOURCE_DIR}/include")
target_link_libraries(server 
    gRPC::grpc++
    protobuf::libprotobuf)
 
# Клиентское приложение
add_executable(client 
    src/client/client.cc 
    ${proto_srcs} 
    ${grpc_srcs})
target_include_directories(client PRIVATE "${generated_dir}" "${CMAKE_SOURCE_DIR}/include")
target_link_libraries(client 
    gRPC::grpc++
    protobuf::libprotobuf)

syntax = "proto3";
 
package ecommerce;
 
service ProductService {
    // Получить информацию о товаре по ID
    rpc GetProduct(GetProductRequest) returns (Product);
    
    // Создать новый товар
    rpc CreateProduct(CreateProductRequest) returns (Product);
    
    // Поиск товаров с потоковым возвратом результатов
    rpc SearchProducts(SearchProductsRequest) returns (stream Product);
    
    // Обновление цен пакетом (клиентский стриминг)
    rpc UpdatePrices(stream UpdatePriceRequest) returns (BatchUpdateResponse);
    
    // Наблюдение за изменениями товаров (двунаправленный стриминг)
    rpc WatchProducts(stream WatchRequest) returns (stream ProductChange);
}
 
message GetProductRequest {
    string product_id = 1;
}
 
message Product {
    string id = 1;
    string name = 2;
    string description = 3;
    double price = 4;
    int32 stock_quantity = 5;
    repeated string categories = 6;
}
 
message CreateProductRequest {
    Product product = 1;
}
 
message SearchProductsRequest {
    string query = 1;
    repeated string categories = 2;
    int32 page_size = 3;
    string page_token = 4;
}
 
message UpdatePriceRequest {
    string product_id = 1;
    double new_price = 2;
}
 
message BatchUpdateResponse {
    int32 success_count = 1;
    int32 failure_count = 2;
}
 
message WatchRequest {
    repeated string product_ids = 1;
    repeated string categories = 2;
}
 
message ProductChange {
    enum ChangeType {
        UNKNOWN = 0;
        CREATED = 1;
        UPDATED = 2;
        DELETED = 3;
    }
    
    ChangeType type = 1;
    Product product = 2;
    string timestamp = 3;
}

#include <iostream>
#include <memory>
#include <string>
#include <map>
#include <mutex>
#include <grpcpp/grpcpp.h>
#include "service.grpc.pb.h"
 
class ProductServiceImpl final : public ecommerce::ProductService::Service {
private:
    std::map<std::string, ecommerce::Product> products;
    std::mutex products_mutex;
 
public:
    grpc::Status GetProduct(grpc::ServerContext* context,
                         const ecommerce::GetProductRequest* request,
                         ecommerce::Product* response) override {
        std::lock_guard<std::mutex> lock(products_mutex);
        
        auto it = products.find(request->product_id());
        if (it == products.end()) {
            return grpc::Status(grpc::StatusCode::NOT_FOUND, 
                              "Товар не найден");
        }
        
        *response = it->second;
        return grpc::Status::OK;
    }
    
    grpc::Status CreateProduct(grpc::ServerContext* context,
                            const ecommerce::CreateProductRequest* request,
                            ecommerce::Product* response) override {
        std::lock_guard<std::mutex> lock(products_mutex);
        
        const auto& product = request->product();
        
        // Проверка входных данных
        if (product.name().empty()) {
            return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, 
                              "Имя товара не может быть пустым");
        }
        
        if (product.price() < 0) {
            return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, 
                              "Цена не может быть отрицательной");
        }
        
        // В реальном сервисе мы бы генерировали ID
        std::string product_id = product.id().empty() 
            ? "prod-" + std::to_string(products.size() + 1) 
            : product.id();
        
        ecommerce::Product new_product = product;
        new_product.set_id(product_id);
        
        products[product_id] = new_product;
        *response = new_product;
        
        return grpc::Status::OK;
    }
    
    grpc::Status SearchProducts(grpc::ServerContext* context,
                             const ecommerce::SearchProductsRequest* request,
                             grpc::ServerWriter<ecommerce::Product>* writer) override {
        std::lock_guard<std::mutex> lock(products_mutex);
        
        const std::string& query = request->query();
        const auto& categories = request->categories();
        
        for (const auto& pair : products) {
            const auto& product = pair.second;
            
            // Простой поиск по подстроке в имени или описании
            bool matches_query = query.empty() || 
                product.name().find(query) != std::string::npos || 
                product.description().find(query) != std::string::npos;
            
            // Проверка категорий если они указаны
            bool matches_category = categories.empty();
            for (const auto& category : product.categories()) {
                for (const auto& requested_category : categories) {
                    if (category == requested_category) {
                        matches_category = true;
                        break;
                    }
                }
                if (matches_category) break;
            }
            
            if (matches_query && matches_category) {
                writer->Write(product);
            }
        }
        
        return grpc::Status::OK;
    }
    
    // Реализации остальных методов опущены для краткости
};
 
void RunServer() {
    std::string server_address("0.0.0.0:50051");
    ProductServiceImpl service;
    
    grpc::ServerBuilder builder;
    // Слушаем на указанном адресе без SSL/TLS
    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
    // Регистрируем наш сервис
    builder.RegisterService(&service);
    
    // Создаем и запускаем сервер
    std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
    std::cout << "Сервер запущен на " << server_address << std::endl;
    
    // Ждем завершения сервера
    server->Wait();
}
 
int main(int argc, char** argv) {
    RunServer();
    return 0;
}

#include <iostream>
#include <memory>
#include <string>
#include <grpcpp/grpcpp.h>
#include "service.grpc.pb.h"
 
class ProductClient {
private:
  std::unique_ptr<ecommerce::ProductService::Stub> stub_;
 
public:
  ProductClient(std::shared_ptr<grpc::Channel> channel)
      : stub_(ecommerce::ProductService::NewStub(channel)) {}
      
  // Получение товара по ID
  bool GetProduct(const std::string& product_id, ecommerce::Product* product) {
      ecommerce::GetProductRequest request;
      request.set_product_id(product_id);
      
      grpc::ClientContext context;
      grpc::Status status = stub_->GetProduct(&context, request, product);
      
      if (!status.ok()) {
          std::cerr << "Ошибка: " << status.error_message() << std::endl;
          return false;
      }
      return true;
  }
  
  // Создание нового товара
  bool CreateProduct(const ecommerce::Product& product_data, 
                   ecommerce::Product* response) {
      ecommerce::CreateProductRequest request;
      *request.mutable_product() = product_data;
      
      grpc::ClientContext context;
      grpc::Status status = stub_->CreateProduct(&context, request, response);
      
      if (!status.ok()) {
          std::cerr << "Ошибка при создании товара: " 
                   << status.error_message() << std::endl;
          return false;
      }
      return true;
  }
  
  // Поиск товаров (с серверным стримингом)
  bool SearchProducts(const std::string& query, 
                    const std::vector<std::string>& categories) {
      ecommerce::SearchProductsRequest request;
      request.set_query(query);
      
      for (const auto& category : categories) {
          request.add_categories(category);
      }
      
      grpc::ClientContext context;
      std::unique_ptr<grpc::ClientReader<ecommerce::Product>> reader(
          stub_->SearchProducts(&context, request));
          
      ecommerce::Product product;
      int count = 0;
      
      while (reader->Read(&product)) {
          std::cout << "Найден товар: " << product.name() 
                   << ", цена: " << product.price() << std::endl;
          count++;
      }
      
      grpc::Status status = reader->Finish();
      if (!status.ok()) {
          std::cerr << "Ошибка при поиске: " 
                   << status.error_message() << std::endl;
          return false;
      }
      
      std::cout << "Всего найдено товаров: " << count << std::endl;
      return true;
  }
  
  // Пакетное обновление цен (клиентский стриминг)
  bool UpdatePrices(const std::vector<std::pair<std::string, double>>& updates) {
      grpc::ClientContext context;
      ecommerce::BatchUpdateResponse response;
      
      std::unique_ptr<grpc::ClientWriter<ecommerce::UpdatePriceRequest>> writer(
          stub_->UpdatePrices(&context, &response));
          
      for (const auto& update : updates) {
          ecommerce::UpdatePriceRequest request;
          request.set_product_id(update.first);
          request.set_new_price(update.second);
          
          if (!writer->Write(request)) {
              // Если запись не удалась, завершаем поток
              break;
          }
      }
      
      // Завершаем стриминг
      writer->WritesDone();
      grpc::Status status = writer->Finish();
      
      if (!status.ok()) {
          std::cerr << "Ошибка при обновлении цен: " 
                   << status.error_message() << std::endl;
          return false;
      }
      
      std::cout << "Успешно обновлено товаров: " << response.success_count() 
               << ", ошибок: " << response.failure_count() << std::endl;
      return true;
  }
};
 
int main(int argc, char** argv) {
  // Создаем канал к серверу без TLS/SSL
  auto channel = grpc::CreateChannel(
      "localhost:50051", grpc::InsecureChannelCredentials());
  ProductClient client(channel);
  
  // Создаем новый товар
  ecommerce::Product product;
  product.set_name("Смартфон Ultra XYZ");
  product.set_description("Флагманский смартфон с лучшей камерой");
  product.set_price(999.99);
  product.set_stock_quantity(10);
  product.add_categories("Electronics");
  product.add_categories("Smartphones");
  
  ecommerce::Product created_product;
  if (client.CreateProduct(product, &created_product)) {
      std::cout << "Товар успешно создан с ID: " 
               << created_product.id() << std::endl;
  }
  
  // Поиск товаров
  client.SearchProducts("смартфон", {"Electronics"});
  
  return 0;
}

#include <grpcpp/grpcpp.h>
#include <grpcpp/alarm.h>
#include "service.grpc.pb.h"
 
class AsyncProductServiceImpl {
private:
  // Класс для обработки асинхронных запросов
  class CallData {
  public:
    // ... Реализация обработчика запросов ...
  };
  
  std::unique_ptr<grpc::ServerCompletionQueue> cq_;
  ecommerce::ProductService::AsyncService service_;
  std::unique_ptr<grpc::Server> server_;
  std::map<std::string, ecommerce::Product> products_;
  std::mutex products_mutex_;
  
public:
  ~AsyncProductServiceImpl() {
    server_->Shutdown();
    cq_->Shutdown();
  }
  
  void Run() {
    std::string server_address("0.0.0.0:50051");
    
    grpc::ServerBuilder builder;
    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
    builder.RegisterService(&service_);
    
    cq_ = builder.AddCompletionQueue();
    server_ = builder.BuildAndStart();
    std::cout << "Асинхронный сервер запущен на " << server_address << std::endl;
    
    // Запускаем обработчики для каждого типа запросов
    HandleRpcs();
  }
  
private:
  void HandleRpcs() {
    // Создаем обработчики для каждого типа RPC
    new CallData(&service_, cq_.get(), this);
    
    void* tag;
    bool ok;
    
    while (true) {
      // Блокируемся в ожидании события в очереди
      GPR_ASSERT(cq_->Next(&tag, &ok));
      
      if (ok) {
        static_cast<CallData*>(tag)->Proceed();
      }
    }
  }
};

#include <thread>
#include <vector>
#include <grpcpp/grpcpp.h>
#include "service.grpc.pb.h"
 
void RunServer(int num_threads) {
  std::string server_address("0.0.0.0:50051");
  ProductServiceImpl service;
  
  grpc::ServerBuilder builder;
  builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
  builder.RegisterService(&service);
  
  // Создаём несколько очередей для параллельной обработки
  std::vector<std::unique_ptr<grpc::ServerCompletionQueue>> queues;
  for (int i = 0; i < num_threads; i++) {
    queues.push_back(builder.AddCompletionQueue());
  }
  
  auto server = builder.BuildAndStart();
  std::cout << "Сервер запущен на " << server_address << std::endl;
  
  // Запускаем отдельные потоки для каждой очереди
  std::vector<std::thread> threads;
  for (int i = 0; i < num_threads; i++) {
    threads.emplace_back([&queues, i]() {
      // Обработка событий в очереди i
      // ...
    });
  }
  
  // Ждём завершения всех потоков
  for (auto& thread : threads) {
    thread.join();
  }
}

// На стороне сервера
std::string server_address("0.0.0.0:50051");
grpc::ServerBuilder builder;
 
// Загружаем ключи и сертификаты
grpc::SslServerCredentialsOptions ssl_opts;
ssl_opts.pem_root_certs = ReadFile("ca.pem");
grpc::SslServerCredentialsOptions::PemKeyCertPair pkcp;
pkcp.private_key = ReadFile("server.key");
pkcp.cert_chain = ReadFile("server.crt");
ssl_opts.pem_key_cert_pairs.push_back(pkcp);
 
// Используем SSL/TLS для сервера
builder.AddListeningPort(server_address, grpc::SslServerCredentials(ssl_opts));
builder.RegisterService(&service);
auto server = builder.BuildAndStart();
 
// На стороне клиента
grpc::SslCredentialsOptions ssl_opts;
ssl_opts.pem_root_certs = ReadFile("ca.pem");
auto channel = grpc::CreateChannel(
    "localhost:50051", grpc::SslCredentials(ssl_opts));
ProductClient client(channel);

// На стороне клиента
auto call_credentials = grpc::MetadataCredentialsFromPlugin(
    std::unique_ptr<grpc::MetadataCredentialsPlugin>(
        new JwtAuthMetadataProcessor(jwt_token)));
 
auto channel_credentials = grpc::SslCredentials(ssl_opts);
auto combined_credentials = grpc::CompositeChannelCredentials(
    channel_credentials, call_credentials);
 
auto channel = grpc::CreateChannel(
    "localhost:50051", combined_credentials);
    
// В каждом запросе
grpc::ClientContext context;
context.AddMetadata("authorization", "Bearer " + jwt_token);

grpc::Status GetProduct(grpc::ServerContext* context,
                     const ecommerce::GetProductRequest* request,
                     ecommerce::Product* response) override {
  // Извлекаем и проверяем токен
  auto auth_metadata = context->client_metadata().find("authorization");
  if (auth_metadata == context->client_metadata().end()) {
    return grpc::Status(grpc::StatusCode::UNAUTHENTICATED, 
                       "Отсутствует токен авторизации");
  }
  
  std::string auth_header(auth_metadata->second.data(), 
                        auth_metadata->second.size());
  
  // Проверяем токен и извлекаем информацию о пользователе
  if (!verifyJwtToken(auth_header, &user_info)) {
    return grpc::Status(grpc::StatusCode::UNAUTHENTICATED, 
                       "Недействительный токен");
  }
  
  // Проверяем права доступа
  if (!hasPermission(user_info, "products.read")) {
    return grpc::Status(grpc::StatusCode::PERMISSION_DENIED, 
                       "Нет прав на просмотр товаров");
  }
  
  // Продолжаем обработку запроса...
}

class CircuitBreaker {
private:
  enum class State { CLOSED, OPEN, HALF_OPEN };
  State state_ = State::CLOSED;
  std::atomic<int> failure_count_ = 0;
  std::atomic<time_t> reset_time_ = 0;
  const int threshold_ = 5;
  const int reset_timeout_ = 30; // секунд
  
public:
  template <typename Func>
  auto Execute(Func func) {
    if (state_ == State::OPEN) {
      if (std::time(nullptr) > reset_time_) {
        state_ = State::HALF_OPEN;
      } else {
        throw CircuitOpenException("Цепь разорвана");
      }
    }
    
    try {
      auto result = func();
      if (state_ == State::HALF_OPEN) {
        state_ = State::CLOSED;
        failure_count_ = 0;
      }
      return result;
    } catch (const std::exception& e) {
      if (++failure_count_ >= threshold_) {
        state_ = State::OPEN;
        reset_time_ = std::time(nullptr) + reset_timeout_;
      }
      throw;
    }
  }
};
 
// Использование с gRPC
CircuitBreaker breaker;
try {
  auto response = breaker.Execute([&]() {
    ecommerce::Product product;
    grpc::ClientContext context;
    return stub_->GetProduct(&context, request, &product);
  });
  
  if (!response.ok()) {
    // Обработка ошибки gRPC
  }
} catch (const CircuitOpenException& e) {
  // Цепь разорвана, используем резервные механизмы
}

grpc::Status RetryWithBackoff(const std::function<grpc::Status()>& operation,
                           int max_retries = 3,
                           int base_delay_ms = 100) {
  grpc::Status status;
  int retries = 0;
  int delay_ms = base_delay_ms;
  
  do {
    status = operation();
    if (status.ok()) return status;
    
    // Повторяем только для определенных типов ошибок
    if (status.error_code() != grpc::StatusCode::UNAVAILABLE &&
        status.error_code() != grpc::StatusCode::DEADLINE_EXCEEDED) {
      return status;
    }
    
    if (++retries >= max_retries) break;
    
    std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
    delay_ms *= 2; // Экспоненциальное увеличение задержки
  } while (true);
  
  return status;
}

class Bulkhead {
private:
  std::mutex mutex_;
  std::condition_variable cv_;
  int max_concurrent_;
  int current_executing_ = 0;
  
public:
  Bulkhead(int max_concurrent) : max_concurrent_(max_concurrent) {}
  
  template <typename Func>
  auto Execute(Func func) {
    std::unique_lock<std::mutex> lock(mutex_);
    if (current_executing_ >= max_concurrent_) {
      throw BulkheadFullException("Превышено максимальное количество параллельных запросов");
    }
    
    current_executing_++;
    lock.unlock();
    
    try {
      auto result = func();
      lock.lock();
      current_executing_--;
      cv_.notify_one();
      return result;
    } catch (...) {
      lock.lock();
      current_executing_--;
      cv_.notify_one();
      throw;
    }
  }
};

FROM debian:bullseye-slim as builder
 
RUN apt-get update && apt-get install -y \
    build-essential \
    cmake \
    git \
    libgrpc++-dev \
    libprotobuf-dev \
    protobuf-compiler \
    protobuf-compiler-grpc
 
WORKDIR /app
COPY . .
RUN mkdir -p build && cd build && \
    cmake .. && \
    make -j$(nproc)
 
FROM debian:bullseye-slim
 
RUN apt-get update && apt-get install -y \
    libgrpc++1 \
    libprotobuf23 \
    && rm -rf /var/lib/apt/lists/*
 
WORKDIR /app
COPY --from=builder /app/build/server .
EXPOSE 50051
 
CMD ["./server"]

service Health {
  rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
}
 
message HealthCheckRequest {
  string service = 1;
}
 
message HealthCheckResponse {
  enum ServingStatus {
    UNKNOWN = 0;
    SERVING = 1;
    NOT_SERVING = 2;
  }
  ServingStatus status = 1;
}

auto channel = grpc::CreateChannel(
    "dns:///product-service.default.svc.cluster.local:50051",
    grpc::InsecureChannelCredentials());

// Загрузка корневого сертификата
std::string root_cert = ReadFile("ca.pem");
 
// Настройка SSL опций
grpc::SslCredentialsOptions ssl_opts;
ssl_opts.pem_root_certs = root_cert;
 
// Создание защищенного канала
auto channel = grpc::CreateChannel(
    "product-service.example.com:50051",
    grpc::SslCredentials(ssl_opts));

#include <grpcpp/ext/health_check_service_server_builder.h>
#include <grpcpp/grpcpp.h>
#include <grpcpp/health_check_service_interface.h>
 
int main() {
  // Создаем resolver factory для работы с DNS
  auto resolver_factory = std::make_shared<grpc::experimental::DnsResolverFactory>();
  grpc::ResolverRegistry::Builder builder;
  builder.RegisterResolverFactory(resolver_factory);
  builder.Build();
 
  // Создаем канал с балансировкой
  auto channel = grpc::CreateChannel(
      "dns:///product-service.default.svc.cluster.local:50051",
      grpc::InsecureChannelCredentials());
 
  // Дополнительная конфигурация балансировки через параметры канала
  grpc::ChannelArguments args;
  args.SetInt(GRPC_ARG_CLIENT_IDLE_TIMEOUT_MS, 500);
  args.SetInt(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, 2000);
  
  // Создание стаба с учетом всех настроек
  auto stub = ecommerce::ProductService::NewStub(
      grpc::CreateCustomChannel(
          "dns:///product-service.default.svc.cluster.local:50051", 
          grpc::InsecureChannelCredentials(), 
          args));
}

void ProcessBatchRequest(grpc::ServerContext* context,
                       const BatchRequest* request,
                       BatchResponse* response) {
  std::vector<Item> items = ExtractItems(request);
  std::vector<std::future<ProcessingResult>> futures;
  
  // Параллельная обработка элементов с использованием thread pool
  ThreadPool pool(std::thread::hardware_concurrency());
  for (const auto& item : items) {
    futures.push_back(pool.enqueue([item]() {
      return ProcessItem(item);
    }));
  }
  
  // Сбор результатов
  for (auto& future : futures) {
    auto result = future.get();
    if (result.success) {
      response->set_success_count(response->success_count() + 1);
    } else {
      response->set_failure_count(response->failure_count() + 1);
      response->add_error_messages(result.error_message);
    }
  }
}

// Пример интеграции с Redis для хранения состояния
#include <sw/redis++/redis++.h>
 
class ProductRepository {
private:
  sw::redis::Redis redis_;
 
public:
  ProductRepository(const std::string& redis_url)
    : redis_(redis_url) {}
  
  bool SaveProduct(const ecommerce::Product& product) {
    try {
      std::string serialized;
      product.SerializeToString(&serialized);
      redis_.set(product.id(), serialized);
      return true;
    } catch (const sw::redis::Error& e) {
      return false;
    }
  }
  
  std::optional<ecommerce::Product> GetProduct(const std::string& id) {
    try {
      auto serialized = redis_.get(id);
      if (!serialized) return std::nullopt;
      
      ecommerce::Product product;
      if (!product.ParseFromString(*serialized)) {
        return std::nullopt;
      }
      return product;
    } catch (const sw::redis::Error& e) {
      return std::nullopt;
    }
  }
};

class LoggingInterceptor : public grpc::ServerInterceptor {
public:
  grpc::Status Intercept(grpc::ServerContext* context,
                       const void* request,
                       grpc::ServerUnaryInvoker* invoker) override {
    const auto& method = context->method();
    const auto& peer = context->peer();
    
    std::cout << "Начало обработки запроса: " << method 
              << " от " << peer << std::endl;
    
    auto start = std::chrono::high_resolution_clock::now();
    grpc::Status status = invoker->Invoke(context, request);
    auto end = std::chrono::high_resolution_clock::now();
    
    std::cout << "Завершение обработки запроса: " << method 
              << ", статус: " << status.error_code()
              << ", время: " 
              << std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count() 
              << "ms" << std::endl;
    
    return status;
  }
};
 
// Использование при создании сервера
grpc::ServerBuilder builder;
builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
std::vector<std::unique_ptr<grpc::ServerInterceptor>> interceptors;
interceptors.push_back(std::make_unique<LoggingInterceptor>());
builder.RegisterService(std::make_unique<grpc::experimental::InterceptedService>(
  &service, std::move(interceptors)));
auto server = builder.BuildAndStart();

#include <opentelemetry/trace/provider.h>
#include <opentelemetry/exporters/prometheus/prometheus_exporter.h>
#include <opentelemetry/exporters/jaeger/jaeger_exporter.h>
 
// Настройка трассировки
void SetupTracing() {
  auto jaeger_exporter = opentelemetry::exporter::jaeger::JaegerExporterFactory::Create();
  auto processor = opentelemetry::trace::BlockingSpanProcessorFactory::Create(std::move(jaeger_exporter));
  auto provider = opentelemetry::trace::TracerProviderFactory::Create(std::move(processor));
  
  opentelemetry::trace::Provider::SetTracerProvider(std::move(provider));
}
 
// Использование трассировки в коде
void ProcessRequest() {
  auto tracer = opentelemetry::trace::Provider::GetTracerProvider()->GetTracer("product_service");
  auto span = tracer->StartSpan("process_request");
  
  // Устанавливаем span как текущий контекст
  opentelemetry::trace::Scope scope(span);
  
  // Выполняем операции, которые будут трассироваться
  // ...
  
  // Завершаем span
  span->End();
}

#include <nlohmann/json.hpp>
#include <chrono>
#include <iostream>
 
void Log(const std::string& level, const std::string& message, 
       const std::map<std::string, std::string>& fields = {}) {
  nlohmann::json log;
  log["timestamp"] = std::chrono::system_clock::now().time_since_epoch().count();
  log["level"] = level;
  log["message"] = message;
  
  for (const auto& [key, value] : fields) {
    log[key] = value;
  }
  
  std::cout << log.dump() << std::endl;
}
 
// Использование
Log("INFO", "Получен запрос на создание товара", {
  {"product_id", "123"},
  {"user_id", "456"},
  {"request_id", request_id}
});

class CustomAllocator : public google::protobuf::Arena::Allocator {
public:
  void* AllocateAligned(size_t size, size_t alignment) override {
    return aligned_alloc(alignment, size);
  }
  
  void DeallocateAligned(void* ptr, size_t size) override {
    free(ptr);
  }
};
 
// Использование
google::protobuf::ArenaOptions options;
options.allocator = std::make_unique<CustomAllocator>();
google::protobuf::Arena arena(options);

grpc::ChannelArguments args;
args.SetInt(GRPC_ARG_MAX_CONCURRENT_STREAMS, 1000);  // Соединений на один канал
args.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, 10000);      // 10 секунд
args.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 5000);    // 5 секунд

// Выделяем буфер заранее
std::vector<char> buffer(1024);
google::protobuf::io::ArrayOutputStream array_out(buffer.data(), buffer.size());
google::protobuf::io::CodedOutputStream coded_out(&array_out);
 
// Записываем сообщение напрямую
message.SerializeWithCachedSizesToArray(
    reinterpret_cast<uint8_t*>(buffer.data()));

class MockProductService : public ecommerce::ProductService::Service {
public:
  MOCK_METHOD(grpc::Status, GetProduct,
            (grpc::ServerContext*, const ecommerce::GetProductRequest*,
             ecommerce::Product*), (override));
  // Мокаем другие методы...
};
 
TEST(ProductClientTest, GetProduct_Success) {
  // Создаём и настраиваем мок
  MockProductService mock_service;
  EXPECT_CALL(mock_service, GetProduct)
      .WillOnce([](grpc::ServerContext*,
                 const ecommerce::GetProductRequest* request,
                 ecommerce::Product* response) {
        EXPECT_EQ("123", request->product_id());
        response->set_id("123");
        response->set_name("Test Product");
        return grpc::Status::OK;
      });
  
  // Запускаем тестовый сервер с моком
  grpc::ServerBuilder builder;
  builder.RegisterService(&mock_service);
  auto server = builder.BuildAndStart();
  
  // Создаём клиента и проверяем его работу
  auto channel = grpc::CreateChannel("localhost:50051", 
                                    grpc::InsecureChannelCredentials());
  ProductClient client(channel);
  
  ecommerce::Product product;
  ASSERT_TRUE(client.GetProduct("123", &product));
  EXPECT_EQ("Test Product", product.name());
  
  server->Shutdown();
}