M04 — gRPC & Protocol Buffers Deep Dive

// user_service.proto
syntax = "proto3";
package user.v1;

// Go package option — ignored by other languages
option go_package = "github.com/acme/user/v1;userv1";

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

// ── Messages ────────────────────────────────────────────────
message User {
string id = 1; // field number = wire tag
string email = 2;
string username = 3;
Role role = 4; // enum field
google.protobuf.Timestamp created_at = 5;
repeated string tags = 6; // list field
oneof contact { // only one field set at a time
string phone = 7;
string slack = 8;
}
}

enum Role {
ROLE_UNSPECIFIED = 0; // proto3 default; first value MUST be 0
ROLE_USER = 1;
ROLE_ADMIN = 2;
}

message GetUserRequest { string user_id = 1; }
message ListUsersRequest { int32 page_size = 1; string page_token = 2; }
message ListUsersResponse{ repeated User users = 1; string next_page_token = 2; }

// ── Service Definition ──────────────────────────────────────
service UserService {
rpc GetUser (GetUserRequest) returns (User);
rpc ListUsers (ListUsersRequest) returns (ListUsersResponse);
rpc WatchUser (GetUserRequest) returns (stream User);
rpc UploadUsers (stream User) returns (google.protobuf.Empty);
rpc SyncUsers (stream GetUserRequest) returns (stream User);
}

// Always reserve deleted field numbers and names:
message User {
reserved 3, 7; // numbers can't be reused
reserved "old_phone", "legacy_id"; // names can't be reused
string id = 1;
string email = 2;
// field 3 was "username" — reserved above
}

// Encoding the integer 300 as a varint:
// 300 in binary = 0000 0001 0010 1100
// Split into 7-bit groups (LSB first): 010 1100 | 000 0010
// Add continuation bits: 1010 1100 | 0000 0010
// Result bytes: 0xAC 0x02

// Encoding field_number=1, wire_type=0, value=150:
// key = (1 << 3) | 0 = 0x08
// value 150 = 0x96 0x01
// Wire bytes: 08 96 01

uint8_t* encode_varint(uint8_t *buf, uint64_t v) {
while (v > 0x7F) {
*buf++ = (uint8_t)(v | 0x80); // set MSB = more bytes
v >>= 7;
}
*buf++ = (uint8_t)v; // last byte, MSB clear
return buf;
}

const uint8_t* decode_varint(const uint8_t *buf, uint64_t *out) {
uint64_t result = 0;
int shift = 0;
do {
if (shift >= 64) return NULL; // malformed
result |= (uint64_t)(*buf & 0x7F) << shift;
shift += 7;
} while (*buf++ & 0x80);
*out = result;
return buf;
}

// Field number=2 (string "hi"), wire type=2:
// key = (2 << 3) | 2 = 0x12
// length varint = 0x02
// data bytes = 0x68 0x69 ('h', 'i')
// Wire bytes: 12 02 68 69

// Nested message: serialise inner message, prefix its byte length
// User { id="abc", email="x@y.com" } encoded then embedded in:
// GetUserResponse { user=<User bytes> }

// Packed repeated fields (default in proto3 for numeric types):
// [1, 2, 3] as repeated int32 field_number=4:
// key = (4<<3)|2 = 0x22 (wire type 2, not 0!)
// length = 3 bytes
// data: 01 02 03 (three varints packed together)

// gRPC timeout header format: ASCII integer + unit suffix
// grpc-timeout: 200m → 200 milliseconds
// grpc-timeout: 5S → 5 seconds
// grpc-timeout: 100000u → 100 ms in microseconds
// Units: H(hours) M(minutes) S(seconds) m(ms) u(µs) n(ns)

/* Conceptual unary interceptor signature (language-agnostic) */
// invoke(ctx, request, method_info, handler) → (response, error)

// Server unary interceptor: JWT auth check (pseudocode)
func AuthInterceptor(ctx, req, info, handler) (resp, err) {
token := metadata.FromIncomingContext(ctx)["authorization"]
if !validate_jwt(token) {
return nil, status.Error(UNAUTHENTICATED, "invalid token")
}
return handler(ctx_with_claims, req)// pass enriched context
}

// Chain multiple interceptors:
// grpc.ChainUnaryInterceptor(LoggingInterceptor, AuthInterceptor, RateLimitInterceptor)

// grpc/health/v1/health.proto (standard)
service Health {
rpc Check (HealthCheckRequest) returns (HealthCheckResponse);
rpc Watch (HealthCheckRequest) returns (stream HealthCheckResponse);
}
message HealthCheckRequest { string service = 1; }
message HealthCheckResponse {
enum ServingStatus {
UNKNOWN = 0;
SERVING = 1;
NOT_SERVING = 2;
SERVICE_UNKNOWN = 3; // health-check for unknown service name
}
ServingStatus status = 1;
}

// Kubernetes grpc probe (k8s 1.24+):
// livenessProbe:
// grpc:
// port: 50051
// service: "user.v1.UserService"

/* Install: sudo apt install libprotobuf-c-dev protobuf-c-compiler */
/* Generate: protoc --c_out=. user.proto */
/* Generated files: user.pb-c.h, user.pb-c.c */

/* Compile: gcc -o demo demo.c user.pb-c.c -lprotobuf-c */

/* grpc_server.c — unary GetUser over gRPC */
#include <grpc/grpc.h>
#include <grpc/support/log.h>
#include "user.pb-c.h"
#include <string.h>

static void run_server(const char *addr) {
grpc_init();
grpc_server *server = grpc_server_create(NULL, NULL);
grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
grpc_server_register_completion_queue(server, cq, NULL);

grpc_server_credentials *creds = grpc_insecure_server_credentials_create();
grpc_server_add_http2_port(server, addr, creds);
grpc_server_credentials_release(creds);
grpc_server_start(server);
gpr_log(GPR_INFO, "gRPC server listening on %s", addr);

/* ── Event loop ────────────────────────────────────── */
while (1) {
grpc_call *call;
grpc_call_details details;
grpc_metadata_array req_meta;
grpc_call_details_init(&details);
grpc_metadata_array_init(&req_meta);

/* request the next incoming call */
grpc_server_request_call(server, &call, &details, &req_meta, cq, cq, (void*)1);
grpc_event ev = grpc_completion_queue_next(
cq, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
if (ev.type != GRPC_OP_COMPLETE) continue;

gpr_log(GPR_INFO, "RPC: %s", grpc_slice_to_c_string(details.method));

/* receive request message */
grpc_byte_buffer *recv_buf = NULL;
grpc_op recv_ops[1] = ;
grpc_call_start_batch(call, recv_ops, 1, (void*)2, NULL);
grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);

/* decode request protobuf */
grpc_byte_buffer_reader rdr;
grpc_byte_buffer_reader_init(&rdr, recv_buf);
grpc_slice req_slice = grpc_byte_buffer_reader_readall(&rdr);
UserV1__GetUserRequest *req = user__v1__get_user_request__unpack(
NULL, GRPC_SLICE_LENGTH(req_slice),
(const uint8_t*)GRPC_SLICE_START_PTR(req_slice));

/* build response */
UserV1__User resp;
user__v1__user__init(&resp);
resp.id = req ? req->user_id : "unknown";
resp.email = "alice@example.com";

size_t resp_len = user__v1__user__get_packed_size(&resp);
uint8_t *resp_buf = malloc(resp_len);
user__v1__user__pack(&resp, resp_buf);

/* gRPC length-prefix framing: 1 byte flag + 4 bytes length */
uint8_t frame_hdr[5] = {0};
frame_hdr[1] = (resp_len >> 24) & 0xFF;
frame_hdr[2] = (resp_len >> 16) & 0xFF;
frame_hdr[3] = (resp_len >> 8) & 0xFF;
frame_hdr[4] = (resp_len ) & 0xFF;

grpc_slice slices[2] = {
grpc_slice_from_copied_buffer((char*)frame_hdr, 5),
grpc_slice_from_copied_buffer((char*)resp_buf, resp_len)
};
grpc_byte_buffer *send_buf = grpc_raw_byte_buffer_create(slices, 2);

/* send response + trailers */
grpc_metadata trailing_meta[1];
memset(trailing_meta, 0, sizeof(trailing_meta));
grpc_op send_ops[3] = {
{.op = GRPC_OP_SEND_INITIAL_METADATA, .data.send_initial_metadata = {0, NULL}},
{.op = GRPC_OP_SEND_MESSAGE, .data.send_message.send_message = send_buf},
{.op = GRPC_OP_SEND_STATUS_FROM_SERVER, .data.send_status_from_server = {
.trailing_metadata_count = 0,
.status = GRPC_STATUS_OK,
.status_details = &grpc_empty_slice()
}},
};
grpc_call_start_batch(call, send_ops, 3, (void*)3, NULL);
grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);

/* cleanup */
free(resp_buf);
if (req) user__v1__get_user_request__free_unpacked(req, NULL);
grpc_byte_buffer_destroy(recv_buf);
grpc_byte_buffer_destroy(send_buf);
grpc_call_unref(call);
}
}