gRPC and Protocol Buffers Patterns

Proto File Structure

protobuf
1syntax = "proto3";
2
3package orca.environment.v1;
4
5option java_package = "orca.server.grpc.environment";
6option java_multiple_files = true;
7
8import "google/protobuf/timestamp.proto";
9import "google/protobuf/empty.proto";
10
11// Service definition
12service EnvironmentService {
13  // Unary RPCs
14  rpc CreateEnvironment(CreateEnvironmentRequest) returns (Environment);
15  rpc GetEnvironment(GetEnvironmentRequest) returns (Environment);
16  rpc ListEnvironments(ListEnvironmentsRequest) returns (ListEnvironmentsResponse);
17  rpc DeleteEnvironment(DeleteEnvironmentRequest) returns (google.protobuf.Empty);
18
19  // Server streaming
20  rpc WatchEnvironment(WatchEnvironmentRequest) returns (stream EnvironmentEvent);
21
22  // Bidirectional streaming
23  rpc StreamLogs(stream LogRequest) returns (stream LogResponse);
24}
25
26// Messages
27message Environment {
28  string id = 1;
29  string name = 2;
30  EnvironmentStatus status = 3;
31  google.protobuf.Timestamp created_at = 4;
32  google.protobuf.Timestamp updated_at = 5;
33  map<string, string> labels = 6;
34}
35
36enum EnvironmentStatus {
37  ENVIRONMENT_STATUS_UNSPECIFIED = 0;
38  ENVIRONMENT_STATUS_PENDING = 1;
39  ENVIRONMENT_STATUS_RUNNING = 2;
40  ENVIRONMENT_STATUS_STOPPED = 3;
41  ENVIRONMENT_STATUS_FAILED = 4;
42}
43
44message CreateEnvironmentRequest {
45  string name = 1;
46  string description = 2;
47  map<string, string> labels = 3;
48}
49
50message GetEnvironmentRequest {
51  string id = 1;
52}
53
54message ListEnvironmentsRequest {
55  int32 page_size = 1;
56  string page_token = 2;
57  string filter = 3;  // e.g., "status=RUNNING"
58}
59
60message ListEnvironmentsResponse {
61  repeated Environment environments = 1;
62  string next_page_token = 2;
63  int32 total_count = 3;
64}
65
66message DeleteEnvironmentRequest {
67  string id = 1;
68}
69
70message WatchEnvironmentRequest {
71  string id = 1;
72}
73
74message EnvironmentEvent {
75  Environment environment = 1;
76  EventType type = 2;
77  google.protobuf.Timestamp timestamp = 3;
78
79  enum EventType {
80    EVENT_TYPE_UNSPECIFIED = 0;
81    EVENT_TYPE_CREATED = 1;
82    EVENT_TYPE_UPDATED = 2;
83    EVENT_TYPE_DELETED = 3;
84  }
85}

Spring gRPC Server Implementation

kotlin
1@GrpcService
2class EnvironmentGrpcService(
3    private val environmentService: EnvironmentService
4) : EnvironmentServiceGrpcKt.EnvironmentServiceCoroutineImplBase() {
5
6    override suspend fun createEnvironment(
7        request: CreateEnvironmentRequest
8    ): Environment {
9        val result = environmentService.create(request.toDomain())
10        return result.toProto()
11    }
12
13    override suspend fun getEnvironment(
14        request: GetEnvironmentRequest
15    ): Environment {
16        val id = UUID.fromString(request.id)
17        val env = environmentService.findById(id)
18            ?: throw StatusException(Status.NOT_FOUND.withDescription("Environment not found: ${request.id}"))
19        return env.toProto()
20    }
21
22    override suspend fun listEnvironments(
23        request: ListEnvironmentsRequest
24    ): ListEnvironmentsResponse {
25        val page = environmentService.findAll(
26            pageSize = request.pageSize,
27            pageToken = request.pageToken.ifEmpty { null }
28        )
29
30        return ListEnvironmentsResponse.newBuilder()
31            .addAllEnvironments(page.items.map { it.toProto() })
32            .setNextPageToken(page.nextToken ?: "")
33            .setTotalCount(page.total)
34            .build()
35    }
36
37    override fun watchEnvironment(
38        request: WatchEnvironmentRequest
39    ): Flow<EnvironmentEvent> = flow {
40        val id = UUID.fromString(request.id)
41
42        environmentService.watchEvents(id).collect { event ->
43            emit(event.toProto())
44        }
45    }
46}

gRPC Client Implementation

kotlin
1@Component
2class ComputeClient(
3    @GrpcClient("compute")
4    private val stub: ComputeServiceGrpcKt.ComputeServiceCoroutineStub
5) {
6
7    suspend fun createInstance(request: CreateInstanceRequest): Instance {
8        return try {
9            stub.createInstance(request)
10        } catch (e: StatusException) {
11            when (e.status.code) {
12                Status.Code.NOT_FOUND -> throw ResourceNotFoundRestException("Instance template not found")
13                Status.Code.ALREADY_EXISTS -> throw ConflictRestException("Instance already exists")
14                Status.Code.RESOURCE_EXHAUSTED -> throw ServiceUnavailableException("Quota exceeded")
15                else -> throw InternalServerException("Compute service error: ${e.message}")
16            }
17        }
18    }
19
20    suspend fun getInstance(id: String): Instance? {
21        return try {
22            stub.getInstance(GetInstanceRequest.newBuilder().setId(id).build())
23        } catch (e: StatusException) {
24            if (e.status.code == Status.Code.NOT_FOUND) null
25            else throw e
26        }
27    }
28}

Error Handling with gRPC Status

kotlin
1// Mapping domain exceptions to gRPC status
2@GrpcAdvice
3class GrpcExceptionHandler {
4
5    @GrpcExceptionHandler(ResourceNotFoundRestException::class)
6    fun handleNotFound(e: ResourceNotFoundRestException): StatusException =
7        StatusException(
8            Status.NOT_FOUND
9                .withDescription(e.message)
10                .withCause(e)
11        )
12
13    @GrpcExceptionHandler(ValidationRestException::class)
14    fun handleValidation(e: ValidationRestException): StatusException =
15        StatusException(
16            Status.INVALID_ARGUMENT
17                .withDescription(e.message)
18                .withCause(e)
19        )
20
21    @GrpcExceptionHandler(ConflictRestException::class)
22    fun handleConflict(e: ConflictRestException): StatusException =
23        StatusException(
24            Status.ALREADY_EXISTS
25                .withDescription(e.message)
26                .withCause(e)
27        )
28}

Interceptors

kotlin
1// Server interceptor for logging/metrics
2@Component
3class LoggingServerInterceptor : ServerInterceptor {
4
5    override fun <ReqT, RespT> interceptCall(
6        call: ServerCall<ReqT, RespT>,
7        headers: Metadata,
8        next: ServerCallHandler<ReqT, RespT>
9    ): ServerCall.Listener<ReqT> {
10        val method = call.methodDescriptor.fullMethodName
11        val startTime = System.currentTimeMillis()
12
13        logger.info("gRPC call started: $method")
14
15        return object : ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(
16            next.startCall(call, headers)
17        ) {
18            override fun onComplete() {
19                val duration = System.currentTimeMillis() - startTime
20                logger.info("gRPC call completed: $method (${duration}ms)")
21                super.onComplete()
22            }
23        }
24    }
25}
26
27// Client interceptor for auth
28@Component
29class AuthClientInterceptor(
30    private val tokenProvider: TokenProvider
31) : ClientInterceptor {
32
33    override fun <ReqT, RespT> interceptCall(
34        method: MethodDescriptor<ReqT, RespT>,
35        callOptions: CallOptions,
36        next: Channel
37    ): ClientCall<ReqT, RespT> {
38        return object : ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
39            next.newCall(method, callOptions)
40        ) {
41            override fun start(responseListener: Listener<RespT>, headers: Metadata) {
42                headers.put(
43                    Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER),
44                    "Bearer ${tokenProvider.getToken()}"
45                )
46                super.start(responseListener, headers)
47            }
48        }
49    }
50}

Proto to Domain Mapping

kotlin
1// Domain to Proto
2fun orca.server.domain.Environment.toProto(): Environment =
3    Environment.newBuilder()
4        .setId(id.toString())
5        .setName(name)
6        .setStatus(status.toProto())
7        .setCreatedAt(createdAt.toProtoTimestamp())
8        .apply { updatedAt?.let { setUpdatedAt(it.toProtoTimestamp()) } }
9        .putAllLabels(labels)
10        .build()
11
12fun EnvironmentStatus.toProto(): orca.grpc.EnvironmentStatus =
13    when (this) {
14        EnvironmentStatus.PENDING -> orca.grpc.EnvironmentStatus.ENVIRONMENT_STATUS_PENDING
15        EnvironmentStatus.RUNNING -> orca.grpc.EnvironmentStatus.ENVIRONMENT_STATUS_RUNNING
16        EnvironmentStatus.STOPPED -> orca.grpc.EnvironmentStatus.ENVIRONMENT_STATUS_STOPPED
17        EnvironmentStatus.FAILED -> orca.grpc.EnvironmentStatus.ENVIRONMENT_STATUS_FAILED
18    }
19
20fun Instant.toProtoTimestamp(): Timestamp =
21    Timestamp.newBuilder()
22        .setSeconds(epochSecond)
23        .setNanos(nano)
24        .build()
25
26// Proto to Domain
27fun CreateEnvironmentRequest.toDomain() = CreateEnvironmentDomainRequest(
28    name = name,
29    description = description.ifEmpty { null },
30    labels = labelsMap
31)

Streaming Patterns

kotlin
1// Server streaming
2override fun watchEnvironment(request: WatchRequest): Flow<Event> = flow {
3    val channel = eventBus.subscribe(request.id)
4
5    try {
6        for (event in channel) {
7            emit(event.toProto())
8        }
9    } finally {
10        channel.cancel()
11    }
12}
13
14// Bidirectional streaming
15override fun chat(requests: Flow<ChatMessage>): Flow<ChatResponse> = flow {
16    requests.collect { message ->
17        val response = processMessage(message)
18        emit(response)
19    }
20}