grpc

syntax = "proto3";
 
package com.example.order;
 
option java_multiple_files = true;
option java_package = "com.example.order.grpc";
 
service OrderService {
  rpc GetOrder (GetOrderRequest) returns (OrderResponse);
  rpc CreateOrder (CreateOrderRequest) returns (OrderResponse);
  rpc ListOrders (ListOrdersRequest) returns (stream OrderResponse);
}
 
message GetOrderRequest {
  string order_id = 1;
}
 
message CreateOrderRequest {
  string customer_id = 1;
  repeated OrderItem items = 2;
  string shipping_address = 3;
}
 
message OrderItem {
  string product_id = 1;
  int32 quantity = 2;
  double price = 3;
}
 
message OrderResponse {
  string order_id = 1;
  string customer_id = 2;
  repeated OrderItem items = 3;
  OrderStatus status = 4;
  double total_amount = 5;
}
 
enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  PENDING = 1;
  CONFIRMED = 2;
  SHIPPED = 3;
  DELIVERED = 4;
  CANCELLED = 5;
}

{
  "orderId": "ORD-12345",
  "customerId": "CUST-789",
  "items": [
    {
      "productId": "PROD-001",
      "quantity": 2,
      "price": 29.99
    }
  ],
  "status": "CONFIRMED",
  "totalAmount": 59.98
}

syntax = "proto3";
 
// Scalar types
message Example {
  double latitude = 1;       // 64-bit float
  float score = 2;           // 32-bit float
  int32 age = 3;             // Variable-length encoding (số âm tốn 10 byte, dùng sint32 nếu hay âm)
  int64 timestamp = 4;       // Variable-length encoding
  bool is_active = 5;        // Boolean
  string name = 6;           // UTF-8 string
  bytes avatar = 7;          // Arbitrary byte data
}
 
// Collections
message Cart {
  repeated CartItem items = 1;          // List<CartItem>
  map<string, int32> quantities = 2;    // Map<String, Integer>
}
 
// Nested messages
message Address {
  message GeoPoint {
    double lat = 1;
    double lng = 2;
  }
  string street = 1;
  string city = 2;
  GeoPoint location = 3;
}
 
// Oneof — chỉ 1 field được set tại một thời điểm
message Payment {
  oneof method {
    CreditCard credit_card = 1;
    BankTransfer bank_transfer = 2;
    Wallet wallet = 3;
  }
}
 
// Optional — phân biệt "not set" vs "default value"
message Filter {
  optional int32 min_price = 1;  // có thể check hasMinPrice()
  optional int32 max_price = 2;
}
 
// Well-known types
import "google/protobuf/timestamp.proto";
import "google/protobuf/duration.proto";
import "google/protobuf/wrappers.proto";
 
message Event {
  google.protobuf.Timestamp created_at = 1;
  google.protobuf.Duration ttl = 2;
  google.protobuf.StringValue nullable_name = 3;
}

message Order {
  reserved 4, 8;
  reserved "old_status", "legacy";
}

rpc GetOrder (GetOrderRequest) returns (OrderResponse);

rpc ListOrders (ListOrdersRequest) returns (stream OrderResponse);

rpc UploadOrderItems (stream OrderItem) returns (UploadSummary);

rpc OrderChat (stream ChatMessage) returns (stream ChatMessage);

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
 
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.0</version>
    </parent>
 
    <groupId>com.example</groupId>
    <artifactId>order-grpc-service</artifactId>
    <version>1.0.0</version>
 
    <properties>
        <java.version>17</java.version>
        <grpc.version>1.60.0</grpc.version>
        <protobuf.version>3.25.1</protobuf.version>
        <grpc-spring-boot.version>3.0.0.RELEASE</grpc-spring-boot.version>
    </properties>
 
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>net.devh</groupId>
            <artifactId>grpc-server-spring-boot-starter</artifactId>
            <version>${grpc-spring-boot.version}</version>
        </dependency>
        <dependency>
            <groupId>net.devh</groupId>
            <artifactId>grpc-client-spring-boot-starter</artifactId>
            <version>${grpc-spring-boot.version}</version>
        </dependency>
        <dependency>
            <groupId>javax.annotation</groupId>
            <artifactId>javax.annotation-api</artifactId>
            <version>1.3.2</version>
        </dependency>
    </dependencies>
 
    <build>
        <extensions>
            <extension>
                <groupId>kr.motd.maven</groupId>
                <artifactId>os-maven-plugin</artifactId>
                <version>1.7.1</version>
            </extension>
        </extensions>
        <plugins>
            <plugin>
                <groupId>org.xolstice.maven.plugins</groupId>
                <artifactId>protobuf-maven-plugin</artifactId>
                <version>0.6.1</version>
                <configuration>
                    <protocArtifact>
                        com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}
                    </protocArtifact>
                    <pluginId>grpc-java</pluginId>
                    <pluginArtifact>
                        io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}
                    </pluginArtifact>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>compile-custom</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

syntax = "proto3";
 
package com.example.order;
 
option java_multiple_files = true;
option java_package = "com.example.orderservice.grpc";
option java_outer_classname = "OrderServiceProto";
 
// ─── Service Definition ───────────────────────────────────
service OrderService {
  // Unary: tạo order mới
  rpc CreateOrder (CreateOrderRequest) returns (OrderResponse);
 
  // Unary: lấy order theo ID
  rpc GetOrder (GetOrderRequest) returns (OrderResponse);
 
  // Server Streaming: lấy danh sách orders của customer
  rpc ListOrders (ListOrdersRequest) returns (stream OrderResponse);
 
  // Client Streaming: upload nhiều items cùng lúc
  rpc BatchCreateItems (stream OrderItem) returns (BatchCreateResponse);
 
  // Bidirectional Streaming: theo dõi order status real-time
  rpc TrackOrders (stream TrackOrderRequest) returns (stream OrderStatusUpdate);
}
 
// ─── Request Messages ─────────────────────────────────────
message CreateOrderRequest {
  string customer_id = 1;
  repeated OrderItem items = 2;
  string shipping_address = 3;
  optional string coupon_code = 4;
}
 
message GetOrderRequest {
  string order_id = 1;
}
 
message ListOrdersRequest {
  string customer_id = 1;
  int32 page_size = 2;
  string page_token = 3;
  OrderStatus status_filter = 4;
}
 
message TrackOrderRequest {
  string order_id = 1;
}
 
// ─── Response Messages ────────────────────────────────────
message OrderResponse {
  string order_id = 1;
  string customer_id = 2;
  repeated OrderItem items = 3;
  OrderStatus status = 4;
  double total_amount = 5;
  string shipping_address = 6;
  string created_at = 7;
}
 
message BatchCreateResponse {
  int32 total_items = 1;
  int32 success_count = 2;
  int32 failed_count = 3;
}
 
message OrderStatusUpdate {
  string order_id = 1;
  OrderStatus previous_status = 2;
  OrderStatus current_status = 3;
  string updated_at = 4;
  string message = 5;
}
 
// ─── Shared Messages ──────────────────────────────────────
message OrderItem {
  string product_id = 1;
  string product_name = 2;
  int32 quantity = 3;
  double unit_price = 4;
}
 
// ─── Enums ────────────────────────────────────────────────
enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  PENDING = 1;
  CONFIRMED = 2;
  PROCESSING = 3;
  SHIPPED = 4;
  DELIVERED = 5;
  CANCELLED = 6;
}

spring:
  application:
    name: order-grpc-service
 
grpc:
  server:
    port: 9090 # gRPC server port (tách biệt với HTTP)
    enable-keep-alive: true
    keep-alive-time: 30s
    keep-alive-timeout: 5s
    permit-keep-alive-without-calls: true
    max-inbound-message-size: 4MB
 
  client:
    # Config cho khi service này gọi service khác
    inventory-service:
      address: static://localhost:9091
      negotiation-type: plaintext
      enable-keep-alive: true
      keep-alive-without-calls: true
    order-service: # tên dùng trong @GrpcClient("order-service") ở bước 5
      address: static://localhost:9090
      negotiation-type: plaintext
 
logging:
  level:
    com.example: DEBUG
    net.devh.boot.grpc: DEBUG

package com.example.orderservice.service;
 
import com.example.orderservice.grpc.*;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import net.devh.boot.grpc.server.service.GrpcService;
 
import java.time.Instant;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.UUID;
 
/**
 * gRPC Service implementation cho Order Management.
 *
 * Annotation @GrpcService tương đương @Service của Spring,
 * nhưng đăng ký class này như một gRPC service handler.
 * Spring Boot auto-configure sẽ tự động start gRPC server
 * và register service này.
 */
@GrpcService
public class OrderGrpcService extends OrderServiceGrpc.OrderServiceImplBase {
 
    // In-memory store (thay bằng database trong production)
    private final Map<String, OrderResponse> orderStore = new ConcurrentHashMap<>();
 
    // ═══════════════════════════════════════════════════════════
    // UNARY RPC: CreateOrder
    // Client gửi 1 request, server trả 1 response
    // ═══════════════════════════════════════════════════════════
    @Override
    public void createOrder(CreateOrderRequest request,
                            StreamObserver<OrderResponse> responseObserver) {
 
        // Validate input
        if (request.getCustomerId().isBlank()) {
            responseObserver.onError(
                Status.INVALID_ARGUMENT
                    .withDescription("customer_id is required")
                    .asRuntimeException()
            );
            return;
        }
 
        if (request.getItemsList().isEmpty()) {
            responseObserver.onError(
                Status.INVALID_ARGUMENT
                    .withDescription("Order must have at least one item")
                    .asRuntimeException()
            );
            return;
        }
 
        // Calculate total
        double totalAmount = request.getItemsList().stream()
            .mapToDouble(item -> item.getUnitPrice() * item.getQuantity())
            .sum();
 
        // Build response
        String orderId = "ORD-" + UUID.randomUUID().toString().substring(0, 8);
        OrderResponse order = OrderResponse.newBuilder()
            .setOrderId(orderId)
            .setCustomerId(request.getCustomerId())
            .addAllItems(request.getItemsList())
            .setStatus(OrderStatus.PENDING)
            .setTotalAmount(totalAmount)
            .setShippingAddress(request.getShippingAddress())
            .setCreatedAt(Instant.now().toString())
            .build();
 
        // Persist
        orderStore.put(orderId, order);
 
        // Trả response — flow giống REST controller
        // onNext() = gửi response message
        // onCompleted() = kết thúc call (đóng stream của call này, connection HTTP/2 vẫn giữ)
        responseObserver.onNext(order);
        responseObserver.onCompleted();
    }
 
    // ═══════════════════════════════════════════════════════════
    // UNARY RPC: GetOrder
    // ═══════════════════════════════════════════════════════════
    @Override
    public void getOrder(GetOrderRequest request,
                         StreamObserver<OrderResponse> responseObserver) {
 
        String orderId = request.getOrderId();
        OrderResponse order = orderStore.get(orderId);
 
        if (order == null) {
            // gRPC dùng Status codes thay vì HTTP status codes
            // NOT_FOUND tương đương HTTP 404
            responseObserver.onError(
                Status.NOT_FOUND
                    .withDescription("Order not found: " + orderId)
                    .asRuntimeException()
            );
            return;
        }
 
        responseObserver.onNext(order);
        responseObserver.onCompleted();
    }
 
    // ═══════════════════════════════════════════════════════════
    // SERVER STREAMING: ListOrders
    // Client gửi 1 request, server trả về NHIỀU responses
    // ═══════════════════════════════════════════════════════════
    @Override
    public void listOrders(ListOrdersRequest request,
                           StreamObserver<OrderResponse> responseObserver) {
 
        String customerId = request.getCustomerId();
        OrderStatus statusFilter = request.getStatusFilter();
 
        orderStore.values().stream()
            .filter(order -> order.getCustomerId().equals(customerId))
            .filter(order -> statusFilter == OrderStatus.ORDER_STATUS_UNSPECIFIED
                             || order.getStatus() == statusFilter)
            .forEach(order -> {
                // Mỗi lần gọi onNext() = gửi 1 message trong stream
                // Client nhận được từng message một, không cần đợi tất cả
                responseObserver.onNext(order);
 
                // Simulate delay (trong thực tế, data có thể đến từ DB cursor)
                try { Thread.sleep(100); } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
 
        // Signal: stream đã kết thúc
        responseObserver.onCompleted();
    }
 
    // ═══════════════════════════════════════════════════════════
    // CLIENT STREAMING: BatchCreateItems
    // Client gửi NHIỀU requests, server trả 1 response cuối cùng
    // ═══════════════════════════════════════════════════════════
    @Override
    public StreamObserver<OrderItem> batchCreateItems(
            StreamObserver<BatchCreateResponse> responseObserver) {
 
        // Trả về một StreamObserver để NHẬN data từ client
        return new StreamObserver<>() {
            private final List<OrderItem> items = new ArrayList<>();
            private int failedCount = 0;
 
            @Override
            public void onNext(OrderItem item) {
                // Được gọi mỗi khi client gửi 1 item
                if (item.getQuantity() <= 0) {
                    failedCount++;
                } else {
                    items.add(item);
                }
            }
 
            @Override
            public void onError(Throwable t) {
                // Client gặp lỗi hoặc disconnect
                System.err.println("Client error: " + t.getMessage());
            }
 
            @Override
            public void onCompleted() {
                // Client đã gửi xong tất cả items
                // Bây giờ mới trả response
                BatchCreateResponse response = BatchCreateResponse.newBuilder()
                    .setTotalItems(items.size() + failedCount)
                    .setSuccessCount(items.size())
                    .setFailedCount(failedCount)
                    .build();
 
                responseObserver.onNext(response);
                responseObserver.onCompleted();
            }
        };
    }
 
    // ═══════════════════════════════════════════════════════════
    // BIDIRECTIONAL STREAMING: TrackOrders
    // Cả client và server đều gửi stream, độc lập nhau
    // ═══════════════════════════════════════════════════════════
    @Override
    public StreamObserver<TrackOrderRequest> trackOrders(
            StreamObserver<OrderStatusUpdate> responseObserver) {
 
        return new StreamObserver<>() {
            @Override
            public void onNext(TrackOrderRequest request) {
                // Client subscribe theo dõi 1 order
                String orderId = request.getOrderId();
                OrderResponse order = orderStore.get(orderId);
 
                if (order != null) {
                    // Gửi status update ngay lập tức
                    OrderStatusUpdate update = OrderStatusUpdate.newBuilder()
                        .setOrderId(orderId)
                        .setPreviousStatus(OrderStatus.ORDER_STATUS_UNSPECIFIED)
                        .setCurrentStatus(order.getStatus())
                        .setUpdatedAt(Instant.now().toString())
                        .setMessage("Current status for " + orderId)
                        .build();
 
                    responseObserver.onNext(update);
                }
            }
 
            @Override
            public void onError(Throwable t) {
                System.err.println("Client disconnected: " + t.getMessage());
            }
 
            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
    }
}

package com.example.orderservice.client;
 
import com.example.orderservice.grpc.*;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.springframework.stereotype.Service;
 
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
 
@Service
public class OrderGrpcClient {
 
    // @GrpcClient inject một stub tự động từ Spring context
    // "order-service" map với config trong application.yml
    @GrpcClient("order-service")
    private OrderServiceGrpc.OrderServiceBlockingStub blockingStub;
 
    @GrpcClient("order-service")
    private OrderServiceGrpc.OrderServiceStub asyncStub;
 
    // ─── Unary Call (Blocking) ────────────────────────────────
    public OrderResponse createOrder(String customerId,
                                     List<OrderItem> items,
                                     String address) {
        CreateOrderRequest request = CreateOrderRequest.newBuilder()
            .setCustomerId(customerId)
            .addAllItems(items)
            .setShippingAddress(address)
            .build();
 
        try {
            // Blocking call — thread chờ cho đến khi có response (tối đa 3 giây)
            return blockingStub
                .withDeadlineAfter(3, TimeUnit.SECONDS)
                .createOrder(request);
        } catch (StatusRuntimeException e) {
            System.err.println("gRPC error: " + e.getStatus());
            throw e;
        }
    }
 
    // ─── Unary Call (Blocking) ────────────────────────────────
    public OrderResponse getOrder(String orderId) {
        GetOrderRequest request = GetOrderRequest.newBuilder()
            .setOrderId(orderId)
            .build();
 
        try {
            return blockingStub
                .withDeadlineAfter(3, TimeUnit.SECONDS)
                .getOrder(request);
        } catch (StatusRuntimeException e) {
            if (e.getStatus().getCode() == io.grpc.Status.Code.NOT_FOUND) {
                System.err.println("Order not found: " + orderId);
                return null;
            }
            throw e;
        }
    }
 
    // ─── Server Streaming (Blocking Iterator) ─────────────────
    public List<OrderResponse> listOrders(String customerId) {
        ListOrdersRequest request = ListOrdersRequest.newBuilder()
            .setCustomerId(customerId)
            .build();
 
        List<OrderResponse> orders = new ArrayList<>();
 
        // blockingStub trả về Iterator cho server streaming
        // Mỗi lần gọi next() sẽ block cho đến khi server gửi message tiếp
        Iterator<OrderResponse> iterator = blockingStub.listOrders(request);
        while (iterator.hasNext()) {
            OrderResponse order = iterator.next();
            orders.add(order);
            System.out.println("Received order: " + order.getOrderId());
        }
 
        return orders;
    }
 
    // ─── Server Streaming (Async) ─────────────────────────────
    public void listOrdersAsync(String customerId) throws InterruptedException {
        ListOrdersRequest request = ListOrdersRequest.newBuilder()
            .setCustomerId(customerId)
            .build();
 
        CountDownLatch latch = new CountDownLatch(1);
 
        // Async stub — non-blocking, callback-based
        asyncStub.listOrders(request, new StreamObserver<>() {
            @Override
            public void onNext(OrderResponse order) {
                System.out.println("Async received: " + order.getOrderId());
            }
 
            @Override
            public void onError(Throwable t) {
                System.err.println("Stream error: " + t.getMessage());
                latch.countDown();
            }
 
            @Override
            public void onCompleted() {
                System.out.println("Stream completed");
                latch.countDown();
            }
        });
 
        // Đợi stream hoàn thành (max 30 giây)
        latch.await(30, TimeUnit.SECONDS);
    }
 
    // ─── Client Streaming (Async) ─────────────────────────────
    public void batchCreateItems(List<OrderItem> items) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
 
        // asyncStub trả về StreamObserver để GỬI data lên server
        StreamObserver<OrderItem> requestObserver = asyncStub.batchCreateItems(
            new StreamObserver<>() {
                @Override
                public void onNext(BatchCreateResponse response) {
                    System.out.printf("Batch result: %d/%d succeeded%n",
                        response.getSuccessCount(),
                        response.getTotalItems());
                }
 
                @Override
                public void onError(Throwable t) {
                    latch.countDown();
                }
 
                @Override
                public void onCompleted() {
                    latch.countDown();
                }
            }
        );
 
        // Gửi từng item lên server
        for (OrderItem item : items) {
            requestObserver.onNext(item);
            Thread.sleep(50); // Simulate delay
        }
 
        // Signal: đã gửi xong
        requestObserver.onCompleted();
        latch.await(10, TimeUnit.SECONDS);
    }
}

package com.example.orderservice;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class OrderGrpcServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(OrderGrpcServiceApplication.class, args);
        // gRPC server tự động start trên port 9090
        // Spring Boot web server (nếu có) chạy trên port 8080
    }
}

mvn spring-boot:run

import com.google.rpc.ErrorInfo;
import com.google.rpc.BadRequest;
import io.grpc.protobuf.StatusProto;
 
// ─── Trả về error với metadata chi tiết ───────────────────
private void handleValidationError(
        StreamObserver<?> responseObserver,
        String field,
        String description) {
 
    // Tạo rich error với field violations
    BadRequest badRequest = BadRequest.newBuilder()
        .addFieldViolations(
            BadRequest.FieldViolation.newBuilder()
                .setField(field)
                .setDescription(description)
                .build()
        )
        .build();
 
    com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
        .setCode(com.google.rpc.Code.INVALID_ARGUMENT_VALUE)
        .setMessage("Validation failed")
        .addDetails(com.google.protobuf.Any.pack(badRequest))
        .build();
 
    responseObserver.onError(StatusProto.toStatusRuntimeException(status));
}
 
// ─── Client-side: parse rich error ────────────────────────
try {
    OrderResponse order = blockingStub.createOrder(request);
} catch (StatusRuntimeException e) {
    com.google.rpc.Status status = StatusProto.fromThrowable(e);
    if (status != null) {
        for (com.google.protobuf.Any detail : status.getDetailsList()) {
            if (detail.is(BadRequest.class)) {
                // unpack() ném InvalidProtocolBufferException (checked): method bao ngoài phải khai báo/bắt
                BadRequest badRequest = detail.unpack(BadRequest.class);
                badRequest.getFieldViolationsList().forEach(violation ->
                    System.err.printf("Field '%s': %s%n",
                        violation.getField(),
                        violation.getDescription())
                );
            }
        }
    }
}

Client (deadline: 5s)
  -> Service A (con 4.8s)
    -> Service B (con 4.2s)
      -> Service C (con 3.5s)
        -> Database (con 2.8s, phai tu dat query timeout)

// ─── Client: set deadline ─────────────────────────────────
OrderResponse order = blockingStub
    .withDeadlineAfter(5, TimeUnit.SECONDS)  // Timeout 5 giây
    .getOrder(request);
 
// ─── Server: check deadline ───────────────────────────────
@Override
public void getOrder(GetOrderRequest request,
                     StreamObserver<OrderResponse> responseObserver) {
 
    // Check xem deadline đã hết chưa trước khi làm việc nặng
    if (Context.current().isCancelled()) {
        responseObserver.onError(
            Status.CANCELLED
                .withDescription("Client cancelled or deadline exceeded")
                .asRuntimeException()
        );
        return;
    }
 
    // Nếu service này gọi service khác (trên cùng thread/Context), deadline
    // tự động propagate qua gRPC context — không cần truyền manual
    OrderResponse order = orderStore.get(request.getOrderId());
    if (order == null) {
        responseObserver.onError(Status.NOT_FOUND.asRuntimeException());
        return;
    }
    responseObserver.onNext(order);
    responseObserver.onCompleted();
}

package com.example.orderservice.interceptor;
 
import io.grpc.*;
import net.devh.boot.grpc.server.interceptor.GrpcGlobalServerInterceptor;
 
@GrpcGlobalServerInterceptor  // Áp dụng cho TẤT CẢ gRPC services
public class LoggingInterceptor implements ServerInterceptor {
 
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call,
            Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {
 
        String methodName = call.getMethodDescriptor().getFullMethodName();
        long startTime = System.currentTimeMillis();
 
        System.out.printf("[gRPC] → %s started%n", methodName);
 
        // Wrap ServerCall để intercept response
        ServerCall<ReqT, RespT> wrappedCall = new ForwardingServerCall
                .SimpleForwardingServerCall<>(call) {
 
            @Override
            public void close(Status status, Metadata trailers) {
                long duration = System.currentTimeMillis() - startTime;
                System.out.printf("[gRPC] ← %s completed | status=%s | %dms%n",
                    methodName, status.getCode(), duration);
                super.close(status, trailers);
            }
        };
 
        return next.startCall(wrappedCall, headers);
    }
}

package com.example.orderservice.interceptor;
 
import io.grpc.*;
import net.devh.boot.grpc.server.interceptor.GrpcGlobalServerInterceptor;
 
@GrpcGlobalServerInterceptor
public class AuthInterceptor implements ServerInterceptor {
 
    // Metadata key cho auth token (tương đương HTTP header)
    private static final Metadata.Key<String> AUTH_KEY =
        Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
 
    // Context key để truyền user info xuống service layer
    public static final Context.Key<String> USER_ID_KEY =
        Context.key("userId");
 
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call,
            Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {
 
        String methodName = call.getMethodDescriptor().getFullMethodName();
 
        // Skip auth cho health check (so khớp đúng service chuẩn, đừng dùng contains("Health"):
        // method nào tình cờ có chữ Health trong tên cũng được bỏ qua auth)
        if (methodName.startsWith("grpc.health.v1.Health/")) {
            return next.startCall(call, headers);
        }
 
        // Extract token từ metadata
        String token = headers.get(AUTH_KEY);
        if (token == null || !token.startsWith("Bearer ")) {
            call.close(
                Status.UNAUTHENTICATED.withDescription("Missing or invalid token"),
                new Metadata()
            );
            return new ServerCall.Listener<>() {};
        }
 
        // Validate token (simplified)
        String userId = validateToken(token.substring(7));
        if (userId == null) {
            call.close(
                Status.UNAUTHENTICATED.withDescription("Invalid token"),
                new Metadata()
            );
            return new ServerCall.Listener<>() {};
        }
 
        // Attach userId vào gRPC Context — service có thể đọc bằng
        // AuthInterceptor.USER_ID_KEY.get()
        Context context = Context.current().withValue(USER_ID_KEY, userId);
        return Contexts.interceptCall(context, call, headers, next);
    }
 
    private String validateToken(String token) {
        // Implement JWT validation logic here
        // Return userId if valid, null if invalid
        return "user-123"; // Simplified
    }
}

package com.example.orderservice.interceptor;
 
import io.grpc.*;
import net.devh.boot.grpc.client.interceptor.GrpcGlobalClientInterceptor;
 
@GrpcGlobalClientInterceptor
public class ClientAuthInterceptor implements ClientInterceptor {
 
    private static final Metadata.Key<String> AUTH_KEY =
        Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
 
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
            MethodDescriptor<ReqT, RespT> method,
            CallOptions callOptions,
            Channel next) {
 
        return new ForwardingClientCall.SimpleForwardingClientCall<>(
                next.newCall(method, callOptions)) {
 
            @Override
            public void start(Listener<RespT> responseListener, Metadata headers) {
                // Attach token vào mọi outgoing request
                headers.put(AUTH_KEY, "Bearer " + getToken());
                super.start(responseListener, headers);
            }
        };
    }
 
    private String getToken() {
        // Get token from security context, cache, etc.
        return "your-jwt-token-here";
    }
}

@GrpcService
public class OrderGrpcService extends OrderServiceGrpc.OrderServiceImplBase {
 
    @Override
    public void createOrder(CreateOrderRequest request,
                            StreamObserver<OrderResponse> responseObserver) {
 
        // Đọc userId từ Context (được set bởi AuthInterceptor)
        String userId = AuthInterceptor.USER_ID_KEY.get();
        System.out.println("Request from user: " + userId);
 
        // ... business logic
    }
}

service DashboardService {
  // Operator subscribe theo dõi orders, server push updates real-time
  rpc WatchOrders (stream WatchFilter) returns (stream DashboardEvent);
}
 
message WatchFilter {
  string region = 1;           // "us-east", "eu-west"
  OrderStatus status = 2;      // Chỉ theo dõi status cụ thể
  bool include_metrics = 3;    // Có gửi kèm metrics không
}
 
message DashboardEvent {
  oneof event {
    OrderCreated order_created = 1;
    OrderUpdated order_updated = 2;
    MetricsSnapshot metrics = 3;
  }
  string timestamp = 10;
}

@GrpcService
public class DashboardGrpcService
        extends DashboardServiceGrpc.DashboardServiceImplBase {
 
    // Quản lý active subscribers
    private final Set<StreamObserver<DashboardEvent>> subscribers =
        ConcurrentHashMap.newKeySet();
 
    @Override
    public StreamObserver<WatchFilter> watchOrders(
            StreamObserver<DashboardEvent> responseObserver) {
 
        // Thêm subscriber
        subscribers.add(responseObserver);
 
        return new StreamObserver<>() {
            private String currentRegion = "";
 
            @Override
            public void onNext(WatchFilter filter) {
                // Client thay đổi filter — ví dụ chuyển từ "us-east"
                // sang "eu-west" mà KHÔNG cần reconnect
                currentRegion = filter.getRegion();
                System.out.println("Filter updated: region=" + currentRegion);
            }
 
            @Override
            public void onError(Throwable t) {
                subscribers.remove(responseObserver);
            }
 
            @Override
            public void onCompleted() {
                subscribers.remove(responseObserver);
                responseObserver.onCompleted();
            }
        };
    }
 
    // Được gọi khi có order mới (từ event bus, Kafka, etc.)
    public void broadcastOrderCreated(OrderCreated event) {
        DashboardEvent dashboardEvent = DashboardEvent.newBuilder()
            .setOrderCreated(event)
            .setTimestamp(Instant.now().toString())
            .build();
 
        // Fan-out tới tất cả subscribers.
        // StreamObserver KHÔNG thread-safe: nếu broadcast được gọi từ nhiều thread
        // (vd nhiều Kafka consumer) thì phải đồng bộ theo từng subscriber.
        subscribers.forEach(subscriber -> {
            try {
                synchronized (subscriber) {
                    subscriber.onNext(dashboardEvent);
                }
            } catch (Exception e) {
                subscribers.remove(subscriber);
            }
        });
    }
}

// Server-side: respect backpressure
@Override
public void listOrders(ListOrdersRequest request,
                       StreamObserver<OrderResponse> responseObserver) {
 
    // Cast để access flow control methods
    ServerCallStreamObserver<OrderResponse> serverObserver =
        (ServerCallStreamObserver<OrderResponse>) responseObserver;
 
    // Handler có thể được gọi lại sau khi đã gửi xong, nên cần cờ để
    // không gọi onCompleted() hai lần (lần hai ném IllegalStateException)
    AtomicBoolean done = new AtomicBoolean(false);
 
    // Callback khi client sẵn sàng nhận thêm data
    serverObserver.setOnReadyHandler(() -> {
        while (!done.get() && serverObserver.isReady()) {
            OrderResponse order = getNextOrder(); // từ DB cursor
            if (order == null) {
                done.set(true);
                serverObserver.onCompleted();
                return;
            }
            serverObserver.onNext(order);
        }
        // Nếu client chưa ready, dừng gửi — tự động resume khi ready
    });
}

// grpc-spring-boot-starter tự động register health service chuẩn
// Client hoặc load balancer có thể check:
// grpc.health.v1.Health/Check
// Lưu ý: HealthIndicator của Actuator dưới đây chỉ hiện ở /actuator/health;
// muốn nó quyết định luôn trạng thái gRPC health thì phải cấu hình thêm
// (vd grpc.server.health-service.type=ACTUATOR ở các bản starter mới)
 
// Custom health indicator
@Component
public class OrderServiceHealthIndicator
        extends AbstractHealthIndicator {
 
    @Override
    protected void doHealthCheck(Health.Builder builder) {
        // Check database connection, dependencies, etc.
        if (isDatabaseHealthy()) {
            builder.up().withDetail("database", "connected");
        } else {
            builder.down().withDetail("database", "disconnected");
        }
    }
}

// Client-side load balancing với round-robin
ManagedChannel channel = ManagedChannelBuilder
    .forTarget("dns:///order-service:9090")
    .defaultLoadBalancingPolicy("round_robin")
    .usePlaintext()
    .build();

// Cấu hình retry trong application.yml
// hoặc programmatically:
Map<String, Object> retryPolicy = Map.of(
    "maxAttempts", 3.0,
    "initialBackoff", "0.5s",
    "maxBackoff", "30s",
    "backoffMultiplier", 2.0,
    // Chỉ retry lỗi chắc chắn là tạm thời. DEADLINE_EXCEEDED thường không nên:
    // deadline áp cho cả lời gọi (mọi lần thử), và request có thể đã chạy xong ở server
    "retryableStatusCodes", List.of("UNAVAILABLE")
);
 
Map<String, Object> serviceConfig = Map.of(
    "methodConfig", List.of(Map.of(
        "name", List.of(Map.of(
            "service", "com.example.order.OrderService"
        )),
        "retryPolicy", retryPolicy
    ))
);
 
ManagedChannel channel = ManagedChannelBuilder
    .forTarget("order-service:9090")
    .defaultServiceConfig(serviceConfig)
    .enableRetry()
    .build();

// Tích hợp với Micrometer (Spring Boot Actuator)
// và OpenTelemetry cho distributed tracing
 
@Configuration
public class GrpcObservabilityConfig {
 
    @Bean
    @GrpcGlobalServerInterceptor // là annotation, không phải kiểu trả về
    public ServerInterceptor metricsInterceptor(
            MeterRegistry meterRegistry) {
        return new ServerInterceptor() {
            @Override
            public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
                    ServerCall<ReqT, RespT> call,
                    Metadata headers,
                    ServerCallHandler<ReqT, RespT> next) {
 
                String method = call.getMethodDescriptor().getBareMethodName();
                Timer.Sample sample = Timer.start(meterRegistry);
 
                return next.startCall(
                    new ForwardingServerCall.SimpleForwardingServerCall<>(call) {
                        @Override
                        public void close(Status status, Metadata trailers) {
                            sample.stop(Timer.builder("grpc.server.calls")
                                .tag("method", method)
                                .tag("status", status.getCode().name())
                                .register(meterRegistry));
                            super.close(status, trailers);
                        }
                    }, headers);
            }
        };
    }
}