gRPC dan Protocol Buffers

// file: user.proto
syntax = "proto3";

package user;

option go_package = "github.com/myapp/proto/user";
option java_package = "com.myapp.proto";

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

// Message definition
message User {
  int32 id = 1;                    // Field number 1
  string name = 2;                 // Field number 2
  string email = 3;                // Field number 3
  UserRole role = 4;               // Enum field
  repeated string tags = 5;        // Repeated = array/list
  map<string, string> metadata = 6; // Map type
  google.protobuf.Timestamp created_at = 7;
  oneof contact {                  // OneOf: pilih salah satu
    string phone = 8;
    string telegram = 9;
  }
  Address address = 10;           // Nested message
}

message Address {
  string street = 1;
  string city = 2;
  string country = 3;
  string postal_code = 4;
}

// Enum
enum UserRole {
  USER_ROLE_UNSPECIFIED = 0;  // Default value (harus 0)
  USER_ROLE_MEMBER = 1;
  USER_ROLE_ADMIN = 2;
  USER_ROLE_SUPERADMIN = 3;
}

// Request/Response messages
message GetUserRequest {
  int32 id = 1;
}

message ListUsersRequest {
  int32 page_size = 1;
  string page_token = 2;
  string filter = 3;  // "role:admin AND name:john"
}

message ListUsersResponse {
  repeated User users = 1;
  string next_page_token = 2;
  int32 total_count = 3;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
  UserRole role = 3;
}

message UpdateUserRequest {
  User user = 1;
  google.protobuf.FieldMask update_mask = 2;  // Partial update
}

message DeleteUserRequest {
  int32 id = 1;
}

message Empty {}

# Wire Types dalam Protobuf:
#
# Type 0 (Varint)    : int32, int64, uint32, uint64, sint32, sint64, bool, enum
# Type 1 (64-bit)    : fixed64, sfixed64, double
# Type 2 (Length-del) : string, bytes, embedded messages, repeated fields
# Type 5 (32-bit)    : fixed32, sfixed32, float

# Field number encoding:
# Field 1-15  → 1 byte tag (hemat untuk field yang sering digunakan)
# Field 16-2047 → 2 byte tag
# → Prioritaskan field yang sering diakses ke nomor 1-15!

# DEFAULT VALUES (proto3):
# int32 → 0, string → "", bool → false
# enum → value pertama (harus 0), bytes → empty
# → Default value TIDAK di-serialize → menghemat bandwidth!

# Kompatibilitas backward/forward:
# ✅ Tambah field baru dengan nomor baru
# ✅ Hapus field (field lama tetap bisa di-parse, di-ignore)
# ❌ JANGAN ubah tipe field
# ❌ JANGAN ubah nomor field
# ❌ JANGAN reuse nomor field yang sudah dihapus

// file: user_service.proto
syntax = "proto3";

package user;

import "user.proto";

// Service definition
service UserService {
  // Unary RPC — request-response biasa
  rpc GetUser(GetUserRequest) returns (User);

  // Unary dengan Empty
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);

  rpc CreateUser(CreateUserRequest) returns (User);

  rpc UpdateUser(UpdateUserRequest) returns (User);

  rpc DeleteUser(DeleteUserRequest) returns (Empty);

  // Server streaming — server mengirim banyak response
  rpc WatchUsers(WatchUsersRequest) returns (stream UserEvent);

  // Client streaming — client mengirim banyak request
  rpc BatchCreateUsers(stream CreateUserRequest) returns (BatchCreateResponse);

  // Bidirectional streaming — keduanya stream
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message WatchUsersRequest {
  string filter = 1;
}

message UserEvent {
  enum EventType {
    EVENT_UNSPECIFIED = 0;
    EVENT_CREATED = 1;
    EVENT_UPDATED = 2;
    EVENT_DELETED = 3;
  }
  EventType type = 1;
  User user = 2;
}

message BatchCreateResponse {
  repeated User users = 1;
  int32 success_count = 2;
  int32 error_count = 3;
  repeated string errors = 4;
}

message ChatMessage {
  string user_id = 1;
  string message = 2;
  google.protobuf.Timestamp timestamp = 3;
}

# Install protoc compiler
# macOS
brew install protobuf

# Ubuntu
sudo apt install protobuf-compiler

# Generate Go code
protoc --go_out=. --go-grpc_out=. proto/*.proto

# Generate Python code
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. proto/*.proto

# Generate Node.js code
npx grpc_tools_node_protoc --js_out=import_style=commonjs:. \
  --grpc_out=grpc_js:. proto/*.proto

# Generate Java code
protoc --java_out=. --grpc-java_out=. proto/*.proto

# Generate ke banyak bahasa sekaligus
protoc \
  --go_out=. --go-grpc_out=. \
  --python_out=. --grpc_python_out=. \
  --js_out=import_style=commonjs:. --grpc_out=grpc_js:. \
  -I. proto/*.proto

import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc
from google.protobuf import empty_pb2

class UserServiceServicer(user_pb2_grpc.UserServiceServicer):
    """Implementasi gRPC UserService"""

    def __init__(self):
        self.users = {}  # In-memory storage
        self.next_id = 1

    def GetUser(self, request, context):
        """Unary RPC — ambil user berdasarkan ID"""
        user = self.users.get(request.id)
        if not user:
            context.abort(grpc.StatusCode.NOT_FOUND,
                         f"User {request.id} not found")
        return user

    def CreateUser(self, request, context):
        """Buat user baru"""
        user = user_pb2.User(
            id=self.next_id,
            name=request.name,
            email=request.email,
            role=request.role
        )
        self.users[self.next_id] = user
        self.next_id += 1
        return user

    def ListUsers(self, request, context):
        """List users dengan pagination"""
        page_size = request.page_size or 10
        users_list = list(self.users.values())

        # Simple pagination
        start = int(request.page_token) if request.page_token else 0
        end = start + page_size
        page = users_list[start:end]

        return user_pb2.ListUsersResponse(
            users=page,
            next_page_token=str(end) if end < len(users_list) else "",
            total_count=len(users_list)
        )

    def WatchUsers(self, request, context):
        """Server streaming — kirim event saat user berubah"""
        import time
        # Simulasi: kirim semua user sebagai event CREATED
        for user in self.users.values():
            yield user_pb2.UserEvent(
                type=user_pb2.UserEvent.EVENT_CREATED,
                user=user
            )
            time.sleep(0.1)  # Delay antar event

def serve():
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            ('grpc.max_send_message_length', 50 * 1024 * 1024),
            ('grpc.max_receive_message_length', 50 * 1024 * 1024),
        ]
    )
    user_pb2_grpc.add_UserServiceServicer_to_server(
        UserServiceServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    print("gRPC server running on port 50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

import grpc
from concurrent import futures
import threading
import queue
import user_pb2
import user_pb2_grpc

class ChatServicer(user_pb2_grpc.UserServiceServicer):
    def __init__(self):
        self.chat_queues = {}  # user_id → Queue
        self.lock = threading.Lock()

    def Chat(self, request_iterator, context):
        """Bidirectional streaming chat"""
        # Setup queue untuk client ini
        user_queue = queue.Queue()
        user_id = None

        # Thread untuk menerima message dari client
        def receive_messages():
            nonlocal user_id
            for message in request_iterator:
                user_id = message.user_id
                if user_id not in self.chat_queues:
                    with self.lock:
                        self.chat_queues[user_id] = user_queue

                # Broadcast ke semua user lain
                with self.lock:
                    for uid, q in self.chat_queues.items():
                        if uid != user_id:
                            q.put(message)

        # Start receiver thread
        receiver = threading.Thread(target=receive_messages)
        receiver.start()

        # Kirim message dari queue ke client
        try:
            while context.is_active():
                try:
                    message = user_queue.get(timeout=1.0)
                    yield message
                except queue.Empty:
                    continue
        finally:
            # Cleanup saat client disconnect
            if user_id and user_id in self.chat_queues:
                with self.lock:
                    del self.chat_queues[user_id]

# Client-side bidirectional streaming
def chat_client():
    channel = grpc.insecure_channel('localhost:50051')
    stub = user_pb2_grpc.UserServiceStub(channel)

    # Create message generator
    def send_messages():
        import sys
        for line in sys.stdin:
            yield user_pb2.ChatMessage(
                user_id="user-1",
                message=line.strip()
            )

    # Bidirectional streaming
    responses = stub.Chat(send_messages())
    for response in responses:
        print(f"[{response.user_id}]: {response.message}")

import grpc
import time
import logging
from functools import wraps

logger = logging.getLogger(__name__)

# ===== SERVER INTERCEPTOR =====
class LoggingInterceptor(grpc.ServerInterceptor):
    """Log semua RPC calls"""

    def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method
        start = time.time()

        # Extract metadata
        metadata = dict(handler_call_details.invocation_metadata)
        logger.info(f"→ {method} | metadata: {metadata}")

        # Continue ke handler
        handler = continuation(handler_call_details)

        if handler is None:
            return handler

        # Wrap handler untuk logging response
        if handler.unary_unary:
            return grpc.unary_unary_rpc_method_handler(
                self._wrap_unary(handler.unary_unary, method, start),
                request_deserializer=handler.request_deserializer,
                response_serializer=handler.response_serializer
            )
        return handler

    def _wrap_unary(self, behavior, method, start):
        def wrapper(request, context):
            try:
                response = behavior(request, context)
                elapsed = time.time() - start
                logger.info(f"← {method} OK ({elapsed:.3f}s)")
                return response
            except grpc.RpcError as e:
                elapsed = time.time() - start
                logger.error(f"✗ {method} ERROR: {e.code()} ({elapsed:.3f}s)")
                raise
        return wrapper

# ===== AUTH INTERCEPTOR =====
class AuthInterceptor(grpc.ServerInterceptor):
    """Validasi token di metadata"""

    VALID_TOKENS = {"secret-token-123": "admin", "token-456": "user"}

    def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)

        # Skip auth untuk public methods
        if handler_call_details.method in ['/grpc.health.v1.Health/Check']:
            return continuation(handler_call_details)

        # Validasi token
        token = metadata.get('authorization', '')
        if token not in self.VALID_TOKENS:
            return grpc.unary_unary_rpc_method_handler(
                self._deny_access
            )

        return continuation(handler_call_details)

    def _deny_access(self, request, context):
        context.abort(grpc.StatusCode.UNAUTHENTICATED, "Invalid token")

# ===== CLIENT INTERCEPTOR =====
class ClientMetadataInterceptor(grpc.UnaryUnaryClientInterceptor):
    """Tambahkan metadata ke setiap client request"""

    def __init__(self, token, client_id):
        self.token = token
        self.client_id = client_id

    def intercept_unary_unary(self, continuation, client_call_details, request):
        # Tambahkan metadata
        metadata = list(client_call_details.invocation_metadata or [])
        metadata.append(('authorization', self.token))
        metadata.append(('x-client-id', self.client_id))
        metadata.append(('x-request-id', str(uuid.uuid4())))

        new_details = client_call_details._replace(
            invocation_metadata=metadata
        )
        return continuation(new_details, request)

# Setup server dengan interceptors
server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[AuthInterceptor(), LoggingInterceptor()]
)

# Setup client dengan interceptor
channel = grpc.insecure_channel('localhost:50051')
interceptor = ClientMetadataInterceptor('secret-token-123', 'api-gateway')
channel = grpc.intercept_channel(channel, interceptor)
stub = user_pb2_grpc.UserServiceStub(channel)

# Data yang sama: User object dengan 10 fields

# JSON representation (147 bytes):
{"id":12345,"name":"John Doe","email":"john@example.com",
 "role":"ADMIN","tags":["dev","ops"],"created_at":"2026-01-15T10:30:00Z"}

# Protobuf representation (52 bytes):
# → 65% LEBIH KECIL dari JSON!

# Benchmark (100,000 iterations):
# ┌─────────────────────────┬────────────┬────────────┐
# │ Metric                  │ JSON       │ Protobuf   │
# ├─────────────────────────┼────────────┼────────────┤
# │ Serialize time          │ 45ms       │ 12ms       │
# │ Deserialize time        │ 52ms       │ 15ms       │
# │ Serialized size         │ 147 bytes  │ 52 bytes   │
# │ Throughput              │ 50K/s      │ 180K/s     │
# └─────────────────────────┴────────────┴────────────┘

# gRPC vs REST Benchmark:
# ┌─────────────────────────┬────────────┬────────────┐
# │ Scenario                │ REST       │ gRPC       │
# ├─────────────────────────┼────────────┼────────────┤
# │ Simple GET (1 record)   │ 2.3ms      │ 0.5ms      │
# │ List 1000 records       │ 45ms       │ 8ms        │
# │ Upload 1MB file         │ 120ms      │ 35ms       │
# │ 10K concurrent clients  │ 12K/s      │ 45K/s      │
# │ Streaming 1000 msg      │ N/A (REST) │ 15ms total │
# └─────────────────────────┴────────────┴────────────┘