gRPC Reference: Protobuf, Go/Python Servers, Streaming & gRPC-Gateway

// users.proto — service definition:
syntax = "proto3";
package users.v1;
option go_package = "github.com/myorg/api/users/v1";

// Request/response messages:
message GetUserRequest {
  string user_id = 1;               // field number 1 (binary wire format tag)
}

message User {
  string id = 1;
  string name = 2;
  string email = 3;
  int32 age = 4;
  repeated string tags = 5;         // repeated = array
  optional string phone = 6;        // optional = may be absent
  map<string, string> metadata = 7; // map field
  UserStatus status = 8;
}

enum UserStatus {
  USER_STATUS_UNSPECIFIED = 0;      // always include 0 as default
  USER_STATUS_ACTIVE = 1;
  USER_STATUS_INACTIVE = 2;
}

// Service definition:
service UserService {
  rpc GetUser(GetUserRequest) returns (User);
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
  rpc CreateUser(CreateUserRequest) returns (User);
  rpc WatchUsers(WatchUsersRequest) returns (stream UserEvent);    // server streaming
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);       // bidirectional
}

# Generate Go stubs:
protoc --go_out=. --go-grpc_out=. users.proto

# Generate Python stubs:
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. users.proto

# Generate TypeScript (with ts-proto):
protoc --plugin=./node_modules/.bin/protoc-gen-ts_proto \
  --ts_proto_out=. users.proto

package main

import (
    "context"
    "log"
    "net"
    usersv1 "github.com/myorg/api/users/v1"
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
)

// Server implementation (implements the generated interface):
type UserServer struct {
    usersv1.UnimplementedUserServiceServer   // embed for forward-compat
    db *DB
}

func (s *UserServer) GetUser(ctx context.Context, req *usersv1.GetUserRequest) (*usersv1.User, error) {
    user, err := s.db.GetUser(ctx, req.UserId)
    if err != nil {
        return nil, status.Errorf(codes.NotFound, "user %s not found", req.UserId)
    }
    return &usersv1.User{Id: user.ID, Name: user.Name, Email: user.Email}, nil
}

// gRPC error codes (use these, not HTTP-like codes):
// codes.NotFound, codes.InvalidArgument, codes.Unauthenticated
// codes.PermissionDenied, codes.Internal, codes.Unavailable
// codes.AlreadyExists, codes.DeadlineExceeded

// Server with interceptors (middleware):
func main() {
    lis, _ := net.Listen("tcp", ":50051")
    srv := grpc.NewServer(
        grpc.UnaryInterceptor(loggingInterceptor),
        grpc.StreamInterceptor(streamInterceptor),
    )
    usersv1.RegisterUserServiceServer(srv, &UserServer{})
    srv.Serve(lis)
}

func loggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    log.Printf("gRPC call: %s", info.FullMethod)
    resp, err := handler(ctx, req)
    if err != nil {
        log.Printf("error: %v", err)
    }
    return resp, err
}

// Client:
conn, err := grpc.NewClient("localhost:50051",
    grpc.WithTransportCredentials(insecure.NewCredentials()),
)
client := usersv1.NewUserServiceClient(conn)
defer conn.Close()

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

user, err := client.GetUser(ctx, &usersv1.GetUserRequest{UserId: "123"})

# pip install grpcio grpcio-tools grpcio-status

# server.py:
import grpc
from concurrent import futures
import users_pb2
import users_pb2_grpc

class UserServicer(users_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        user = db.get_user(request.user_id)
        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"User {request.user_id} not found")
            return users_pb2.User()
        return users_pb2.User(id=user.id, name=user.name, email=user.email)

    # Server streaming:
    def WatchUsers(self, request, context):
        for event in watch_user_events():
            if context.is_active():         # check if client disconnected
                yield users_pb2.UserEvent(user_id=event.user_id, type=event.type)
            else:
                break

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    users_pb2_grpc.add_UserServiceServicer_to_server(UserServicer(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    server.wait_for_termination()

# Async client (grpcio supports asyncio):
import grpc.aio

async def get_user(user_id: str):
    async with grpc.aio.insecure_channel("localhost:50051") as channel:
        stub = users_pb2_grpc.UserServiceStub(channel)
        try:
            user = await stub.GetUser(
                users_pb2.GetUserRequest(user_id=user_id),
                timeout=5.0
            )
            return user
        except grpc.aio.AioRpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                return None
            raise

// Go — server streaming (server sends many, client reads):
func (s *UserServer) ListUsers(req *usersv1.ListUsersRequest, stream usersv1.UserService_ListUsersServer) error {
    users, _ := s.db.ListAll(stream.Context())
    for _, user := range users {
        if err := stream.Send(&usersv1.User{Id: user.ID, Name: user.Name}); err != nil {
            return err    // client disconnected
        }
    }
    return nil
}

// Go client consuming server stream:
stream, err := client.ListUsers(ctx, &usersv1.ListUsersRequest{})
for {
    user, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        log.Printf("stream error: %v", err)
        break
    }
    fmt.Println(user.Name)
}

// Go — client streaming (client sends many, server reads):
stream, _ := client.BulkCreateUsers(ctx)
for _, u := range usersToCreate {
    stream.Send(&usersv1.CreateUserRequest{Name: u.Name, Email: u.Email})
}
response, _ := stream.CloseAndRecv()   // close send, get single response

// Go — bidirectional streaming:
stream, _ := client.Chat(ctx)
go func() {
    for {
        msg, err := stream.Recv()
        if err != nil { break }
        fmt.Println(msg.Text)
    }
}()
for _, msg := range messages {
    stream.Send(&usersv1.ChatMessage{Text: msg})
}
stream.CloseSend()

// Cancellation (client-side timeout or cancel):
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// Or: ctx, cancel := context.WithCancel(context.Background())
// stream.Context().Done() fires when client cancels

// TLS (production — always use in prod):
creds, _ := credentials.NewServerTLSFromFile("cert.pem", "key.pem")
srv := grpc.NewServer(grpc.Creds(creds))

// Mutual TLS (client cert required):
tlsConfig := &tls.Config{
    ClientAuth: tls.RequireAndVerifyClientCert,
    ClientCAs:  certPool,
}
creds := credentials.NewTLS(tlsConfig)

// Metadata (like HTTP headers — pass auth tokens, request IDs):
// Client sending metadata:
md := metadata.New(map[string]string{
    "authorization": "Bearer " + token,
    "x-request-id":  requestID,
})
ctx = metadata.NewOutgoingContext(ctx, md)

// Server reading metadata:
md, ok := metadata.FromIncomingContext(ctx)
authHeader := md.Get("authorization")   // returns []string

// Auth interceptor:
func authInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    md, _ := metadata.FromIncomingContext(ctx)
    token := strings.TrimPrefix(md.Get("authorization")[0], "Bearer ")
    if !validateToken(token) {
        return nil, status.Error(codes.Unauthenticated, "invalid token")
    }
    return handler(ctx, req)
}

// gRPC-Gateway (expose REST endpoint on top of gRPC):
// In .proto, add HTTP annotations:
// rpc GetUser(GetUserRequest) returns (User) {
//   option (google.api.http) = { get: "/v1/users/{user_id}" };
// }
// Then: protoc --grpc-gateway_out=. users.proto
// Runs a reverse proxy: HTTP/JSON <--> gRPC

// Debug tools:
// grpcurl -plaintext localhost:50051 list
// grpcurl -plaintext -d '{"user_id":"123"}' localhost:50051 users.v1.UserService/GetUser
// grpcui -plaintext localhost:50051   # browser UI for gRPC