gRPC in Java Microservices

syntax = "proto3";
 
package acme.order.v1;
option java_multiple_files = true;
option java_package = "com.acme.order.grpc";
 
import "google/protobuf/timestamp.proto";
 
service OrderService {
  rpc GetOrder(GetOrderRequest) returns (Order);
  rpc ListOrders(ListOrdersRequest) returns (stream Order);          // server streaming
  rpc ImportOrders(stream Order) returns (ImportSummary);            // client streaming
  rpc WatchOrders(stream WatchRequest) returns (stream OrderEvent);  // bidirectional
}
 
message Order {
  string id = 1;
  string customer_id = 2;
  Money total = 3;
  OrderStatus status = 4;
  repeated OrderLine lines = 5;
  google.protobuf.Timestamp created_at = 6;
 
  // Field 7 held a deprecated 'legacy_total'. Reserving it stops a future
  // developer reusing the number and silently misreading old messages.
  reserved 7;
  reserved "legacy_total";
 
  string channel = 8;   // added later; old clients ignore it
}
 
message Money {
  int64 minor_units = 1;   // never a float for money
  string currency = 2;     // ISO 4217
}
 
enum OrderStatus {
  // proto3 requires a zero value, and it is what an old client sees for an
  // enum value it does not know.
  ORDER_STATUS_UNSPECIFIED = 0;
  ORDER_STATUS_PLACED = 1;
  ORDER_STATUS_PAID = 2;
  ORDER_STATUS_SHIPPED = 3;
  ORDER_STATUS_CANCELLED = 4;
}

@GrpcService
public class OrderGrpcService extends OrderServiceGrpc.OrderServiceImplBase {
 
    private final OrderQueryService orders;
 
    @Override
    public void getOrder(GetOrderRequest request, StreamObserver<Order> observer) {
        try {
            observer.onNext(mapper.toProto(orders.find(request.getId())));
            observer.onCompleted();
        } catch (ResourceNotFoundException ex) {
            // gRPC status codes, not HTTP. NOT_FOUND, PERMISSION_DENIED,
            // INVALID_ARGUMENT, DEADLINE_EXCEEDED, UNAVAILABLE.
            observer.onError(Status.NOT_FOUND
                    .withDescription("no order " + request.getId())
                    .asRuntimeException());
        }
    }
 
    @Override
    public void listOrders(ListOrdersRequest request, StreamObserver<Order> observer) {
        // Streaming keeps memory flat regardless of result size — the whole
        // point compared to returning a large list.
        try (Stream<OrderEntity> stream = orders.streamByCustomer(request.getCustomerId())) {
            stream.map(mapper::toProto).forEach(observer::onNext);
            observer.onCompleted();
        }
    }
}

@Service
public class OrderClient {
 
    @GrpcClient("order-service")
    private OrderServiceGrpc.OrderServiceBlockingStub stub;
 
    public Order fetch(String id) {
        return stub
                // Without a deadline, a hung server holds this call forever.
                // The deadline propagates to downstream calls automatically.
                .withDeadlineAfter(2, TimeUnit.SECONDS)
                .getOrder(GetOrderRequest.newBuilder().setId(id).build());
    }
}

@GrpcGlobalServerInterceptor
public class AuthInterceptor implements ServerInterceptor {
 
    private static final Metadata.Key<String> AUTHORIZATION =
            Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
 
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
 
        String header = headers.get(AUTHORIZATION);
        if (header == null || !header.startsWith("Bearer ")) {
            call.close(Status.UNAUTHENTICATED.withDescription("missing token"), new Metadata());
            return new ServerCall.Listener<>() { };
        }
 
        try {
            Jwt jwt = decoder.decode(header.substring(7));
            Context context = Context.current().withValue(PRINCIPAL, jwt.getSubject());
            return Contexts.interceptCall(context, call, headers, next);
        } catch (JwtException ex) {
            call.close(Status.UNAUTHENTICATED.withDescription("invalid token"), new Metadata());
            return new ServerCall.Listener<>() { };
        }
    }
}

// gRPC has 17 status codes. Map domain exceptions onto them deliberately.
private StatusRuntimeException toStatus(Exception ex) {
    return switch (ex) {
        case ResourceNotFoundException e -> Status.NOT_FOUND.withDescription(e.getMessage()).asRuntimeException();
        case ValidationException e       -> Status.INVALID_ARGUMENT.withDescription(e.getMessage()).asRuntimeException();
        case AccessDeniedException e     -> Status.PERMISSION_DENIED.asRuntimeException();
        case OptimisticLockException e   -> Status.ABORTED.withDescription("conflict").asRuntimeException();
        case RateLimitException e        -> Status.RESOURCE_EXHAUSTED.asRuntimeException();
        default                          -> Status.INTERNAL.asRuntimeException();
    };
}