/generate-grpc-service

Generate gRPC Service

Automatically generate high-performance gRPC services with Protocol Buffer definitions, streaming support, load balancing, and comprehensive service implementations for multiple programming languages.

When to Use This Command

Use /generate-grpc-service when you need to:

• Build high-performance microservices with binary protocol
• Implement real-time bidirectional streaming communication
• Create strongly-typed service contracts across languages
• Build internal services requiring minimal latency
• Support multiple programming languages with single definition
• Implement efficient mobile/IoT communication protocols

DON'T use this when:

• Building browser-based web applications (limited browser support)
• Simple REST APIs suffice (gRPC adds complexity)
• Working with teams unfamiliar with Protocol Buffers
• Debugging tools are limited in your environment

Design Decisions

This command implements gRPC with Protocol Buffers v3 as the primary approach because:

• Binary protocol offers 20-30% better performance than JSON
• Built-in code generation for 10+ languages
• Native support for streaming in all RPC patterns
• Strong typing prevents runtime errors
• Backward compatibility through field numbering
• Built-in service discovery and load balancing

Alternative considered: Apache Thrift

• Similar performance characteristics
• Less ecosystem support
• Fewer language bindings
• Recommended for Facebook ecosystem

Alternative considered: GraphQL with subscriptions

• Better for public APIs
• More flexible queries
• Higher overhead
• Recommended for client-facing APIs

Prerequisites

Before running this command:

1. Protocol Buffer compiler (protoc) installed
2. Language-specific gRPC tools installed
3. Understanding of Protocol Buffer syntax
4. Service architecture defined
5. Authentication strategy determined

Implementation Process

Step 1: Define Service Contract

Create comprehensive .proto files with service definitions and message types.

Step 2: Generate Language Bindings

Compile Protocol Buffers to target language code with gRPC plugins.

Step 3: Implement Service Logic

Build server-side implementations for all RPC methods.

Step 4: Add Interceptors

Implement cross-cutting concerns like auth, logging, and error handling.

Step 5: Configure Production Settings

Set up TLS, connection pooling, and load balancing.

Output Format

The command generates:

• proto/service.proto - Protocol Buffer definitions
• server/ - Server implementation with all RPC methods
• client/ - Client library with connection management
• interceptors/ - Authentication, logging, metrics interceptors
• config/ - TLS certificates and configuration
• docs/api.md - Service documentation

Code Examples

Example 1: E-commerce Service with All RPC Patterns

// proto/ecommerce.proto
syntax = "proto3";

package ecommerce.v1;

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

// Service definition with all RPC patterns
service ProductService {
  // Unary RPC
  rpc GetProduct(GetProductRequest) returns (Product);

  // Server streaming
  rpc ListProducts(ListProductsRequest) returns (stream Product);

  // Client streaming
  rpc ImportProducts(stream Product) returns (ImportSummary);

  // Bidirectional streaming
  rpc WatchInventory(stream InventoryUpdate) returns (stream InventoryChange);

  // Batch operations
  rpc BatchGetProducts(BatchGetProductsRequest) returns (BatchGetProductsResponse);
}

// Message definitions
message Product {
  string id = 1;
  string name = 2;
  string description = 3;
  double price = 4;
  int32 inventory = 5;
  repeated string categories = 6;
  map<string, string> metadata = 7;
  google.protobuf.Timestamp created_at = 8;
  google.protobuf.Timestamp updated_at = 9;

  enum Status {
    STATUS_UNSPECIFIED = 0;
    STATUS_ACTIVE = 1;
    STATUS_DISCONTINUED = 2;
    STATUS_OUT_OF_STOCK = 3;
  }
  Status status = 10;
}

message GetProductRequest {
  string product_id = 1;
  repeated string fields = 2; // Field mask for partial responses
}

message ListProductsRequest {
  string category = 1;
  int32 page_size = 2;
  string page_token = 3;
  string order_by = 4;

  message Filter {
    double min_price = 1;
    double max_price = 2;
    repeated string tags = 3;
  }
  Filter filter = 5;
}

message ImportSummary {
  int32 total_received = 1;
  int32 successful = 2;
  int32 failed = 3;
  repeated ImportError errors = 4;
}

message ImportError {
  int32 index = 1;
  string product_id = 2;
  string error = 3;
}

message InventoryUpdate {
  string product_id = 1;
  int32 quantity_change = 2;
  string warehouse_id = 3;
}

message InventoryChange {
  string product_id = 1;
  int32 old_quantity = 2;
  int32 new_quantity = 3;
  google.protobuf.Timestamp timestamp = 4;
  string triggered_by = 5;
}

message BatchGetProductsRequest {
  repeated string product_ids = 1;
  repeated string fields = 2;
}

message BatchGetProductsResponse {
  repeated Product products = 1;
  repeated string not_found = 2;
}

// server/main.go - Go server implementation
package main

import (
    "context"
    "crypto/tls"
    "fmt"
    "io"
    "log"
    "net"
    "sync"
    "time"

    pb "github.com/company/ecommerce/proto"
    "github.com/golang/protobuf/ptypes/empty"
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/keepalive"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
    "google.golang.org/protobuf/types/known/timestamppb"
)

type productServer struct {
    pb.UnimplementedProductServiceServer
    mu       sync.RWMutex
    products map[string]*pb.Product
    watchers map[string]chan *pb.InventoryChange
}

// Unary RPC implementation
func (s *productServer) GetProduct(
    ctx context.Context,
    req *pb.GetProductRequest,
) (*pb.Product, error) {
    // Extract metadata for tracing
    if md, ok := metadata.FromIncomingContext(ctx); ok {
        if traceID := md.Get("trace-id"); len(traceID) > 0 {
            log.Printf("GetProduct request - trace: %s", traceID[0])
        }
    }

    s.mu.RLock()
    product, exists := s.products[req.ProductId]
    s.mu.RUnlock()

    if !exists {
        return nil, status.Errorf(
            codes.NotFound,
            "product %s not found",
            req.ProductId,
        )
    }

    // Apply field mask if specified
    if len(req.Fields) > 0 {
        return applyFieldMask(product, req.Fields), nil
    }

    return product, nil
}

// Server streaming implementation
func (s *productServer) ListProducts(
    req *pb.ListProductsRequest,
    stream pb.ProductService_ListProductsServer,
) error {
    s.mu.RLock()
    defer s.mu.RUnlock()

    count := 0
    for _, product := range s.products {
        // Apply filters
        if !matchesFilter(product, req) {
            continue
        }

        // Send product to stream
        if err := stream.Send(product); err != nil {
            return status.Errorf(
                codes.Internal,
                "failed to send product: %v",
                err,
            )
        }

        count++
        if req.PageSize > 0 && count >= int(req.PageSize) {
            break
        }

        // Simulate real-time processing
        time.Sleep(10 * time.Millisecond)
    }

    return nil
}

// Client streaming implementation
func (s *productServer) ImportProducts(
    stream pb.ProductService_ImportProductsServer,
) error {
    var summary pb.ImportSummary
    var errors []*pb.ImportError
    index := 0

    for {
        product, err := stream.Recv()
        if err == io.EOF {
            // Client finished sending
            summary.Errors = errors
            return stream.SendAndClose(&summary)
        }
        if err != nil {
            return status.Errorf(
                codes.Internal,
                "failed to receive product: %v",
                err,
            )
        }

        summary.TotalReceived++

        // Validate and store product
        if err := validateProduct(product); err != nil {
            summary.Failed++
            errors = append(errors, &pb.ImportError{
                Index:     int32(index),
                ProductId: product.Id,
                Error:     err.Error(),
            })
        } else {
            s.mu.Lock()
            s.products[product.Id] = product
            s.mu.Unlock()
            summary.Successful++
        }

        index++
    }
}

// Bidirectional streaming implementation
func (s *productServer) WatchInventory(
    stream pb.ProductService_WatchInventoryServer,
) error {
    // Create change channel for this client
    changeChan := make(chan *pb.InventoryChange, 100)
    clientID := generateClientID()

    s.mu.Lock()
    s.watchers[clientID] = changeChan
    s.mu.Unlock()

    defer func() {
        s.mu.Lock()
        delete(s.watchers, clientID)
        s.mu.Unlock()
        close(changeChan)
    }()

    // Handle bidirectional communication
    errChan := make(chan error, 2)

    // Goroutine to receive updates from client
    go func() {
        for {
            update, err := stream.Recv()
            if err == io.EOF {
                errChan <- nil
                return
            }
            if err != nil {
                errChan <- err
                return
            }

            // Process inventory update
            if err := s.processInventoryUpdate(update); err != nil {
                log.Printf("Failed to process update: %v", err)
                continue
            }

            // Notify all watchers
            change := &pb.InventoryChange{
                ProductId:   update.ProductId,
                NewQuantity: s.getInventory(update.ProductId),
                Timestamp:   timestamppb.Now(),
                TriggeredBy: clientID,
            }

            s.broadcastChange(change)
        }
    }()

    // Goroutine to send changes to client
    go func() {
        for change := range changeChan {
            if err := stream.Send(change); err != nil {
                errChan <- err
                return
            }
        }
    }()

    // Wait for error or completion
    return <-errChan
}

// Interceptor for authentication
func authInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    // Extract token from metadata
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        return nil, status.Error(codes.Unauthenticated, "missing metadata")
    }

    tokens := md.Get("authorization")
    if len(tokens) == 0 {
        return nil, status.Error(codes.Unauthenticated, "missing token")
    }

    // Validate token (implement your auth logic)
    if !isValidToken(tokens[0]) {
        return nil, status.Error(codes.Unauthenticated, "invalid token")
    }

    // Continue to handler
    return handler(ctx, req)
}

// Main server setup
func main() {
    // Load TLS credentials
    cert, err := tls.LoadX509KeyPair("server.crt", "server.key")
    if err != nil {
        log.Fatalf("Failed to load certificates: %v", err)
    }

    config := &tls.Config{
        Certificates: []tls.Certificate{cert},
        ClientAuth:   tls.RequireAndVerifyClientCert,
    }

    creds := credentials.NewTLS(config)

    // Configure server options
    opts := []grpc.ServerOption{
        grpc.Creds(creds),
        grpc.UnaryInterceptor(authInterceptor),
        grpc.KeepaliveParams(keepalive.ServerParameters{
            MaxConnectionIdle: 5 * time.Minute,
            Time:             2 * time.Minute,
            Timeout:          20 * time.Second,
        }),
        grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
            MinTime:             5 * time.Second,
            PermitWithoutStream: true,
        }),
        grpc.MaxConcurrentStreams(1000),
    }

    // Create gRPC server
    server := grpc.NewServer(opts...)

    // Register service
    pb.RegisterProductServiceServer(server, &productServer{
        products: make(map[string]*pb.Product),
        watchers: make(map[string]chan *pb.InventoryChange),
    })

    // Start listening
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("Failed to listen: %v", err)
    }

    log.Println("gRPC server starting on :50051")
    if err := server.Serve(lis); err != nil {
        log.Fatalf("Failed to serve: %v", err)
    }
}

Example 2: Python Client with Retry and Load Balancing

# client/product_client.py
import grpc
import asyncio
import logging
from typing import List, Optional, AsyncIterator
from concurrent import futures
from grpc import aio
import backoff

from proto import ecommerce_pb2 as pb
from proto import ecommerce_pb2_grpc as pb_grpc

logger = logging.getLogger(__name__)

class ProductClient:
    """Enhanced gRPC client with retry, load balancing, and connection pooling."""

    def __init__(
        self,
        servers: List[str],
        api_key: Optional[str] = None,
        use_tls: bool = True,
        pool_size: int = 10
    ):
        self.servers = servers
        self.api_key = api_key
        self.use_tls = use_tls
        self.pool_size = pool_size
        self.channels = []
        self.stubs = []
        self._round_robin_counter = 0
        self._setup_channels()

    def _setup_channels(self):
        """Set up connection pool with load balancing."""
        for server in self.servers:
            for _ in range(self.pool_size // len(self.servers)):
                if self.use_tls:
                    # Load client certificates
                    with open('client.crt', 'rb') as f:
                        client_cert = f.read()
                    with open('client.key', 'rb') as f:
                        client_key = f.read()
                    with open('ca.crt', 'rb') as f:
                        ca_cert = f.read()

                    credentials = grpc.ssl_channel_credentials(
                        root_certificates=ca_cert,
                        private_key=client_key,
                        certificate_chain=client_cert
                    )
                    channel = aio.secure_channel(
                        server,
                        credentials,
                        options=[
                            ('grpc.keepalive_time_ms', 120000),
                            ('grpc.keepalive_timeout_ms', 20000),
                            ('grpc.keepalive_permit_without_calls', True),
                            ('grpc.http2.max_pings_without_data', 0),
                        ]
                    )
                else:
                    channel = aio.insecure_channel(
                        server,
                        options=[
                            ('grpc.keepalive_time_ms', 120000),
                            ('grpc.keepalive_timeout_ms', 20000),
                        ]
                    )

                self.channels.append(channel)
                self.stubs.append(pb_grpc.ProductServiceStub(channel))

    def _get_stub(self) -> pb_grpc.ProductServiceStub:
        """Get next stub using round-robin load balancing."""
        stub = self.stubs[self._round_robin_counter]
        self._round_robin_counter = (self._round_robin_counter + 1) % len(self.stubs)
        return stub

    def _get_metadata(self) -> List[tuple]:
        """Generate request metadata."""
        metadata = []
        if self.api_key:
            metadata.append(('authorization', f'Bearer {self.api_key}'))
        metadata.append(('trace-id', self._generate_trace_id()))
        return metadata

    @backoff.on_exception(
        backoff.expo,
        grpc.RpcError,
        max_tries=3,
        giveup=lambda e: e.code() != grpc.StatusCode.UNAVAILABLE
    )
    async def get_product(
        self,
        product_id: str,
        fields: Optional[List[str]] = None
    ) -> pb.Product:
        """Get single product with retry logic."""
        request = pb.GetProductRequest(
            product_id=product_id,
            fields=fields or []
        )

        try:
            response = await self._get_stub().GetProduct(
                request,
                metadata=self._get_metadata(),
                timeout=5.0
            )
            return response
        except grpc.RpcError as e:
            logger.error(f"Failed to get product {product_id}: {e.details()}")
            raise

    async def list_products(
        self,
        category: Optional[str] = None,
        page_size: int = 100,
        min_price: Optional[float] = None,
        max_price: Optional[float] = None
    ) -> AsyncIterator[pb.Product]:
        """Stream products with server-side streaming."""
        request = pb.ListProductsRequest(
            category=category or "",
            page_size=page_size
        )

        if min_price is not None or max_price is not None:
            request.filter.CopyFrom(pb.ListProductsRequest.Filter(
                min_price=min_price or 0,
                max_price=max_price or float('inf')
            ))

        try:
            stream = self._get_stub().ListProducts(
                request,
                metadata=self._get_metadata(),
                timeout=30.0
            )

            async for product in stream:
                yield product
        except grpc.RpcError as e:
            logger.error(f"Failed to list products: {e.details()}")
            raise

    async def import_products(
        self,
        products: List[pb.Product]
    ) -> pb.ImportSummary:
        """Import products using client-side streaming."""
        async def generate_products():
            for product in products:
                yield product
                await asyncio.sleep(0.01)  # Rate limiting

        try:
            response = await self._get_stub().ImportProducts(
                generate_products(),
                metadata=self._get_metadata(),
                timeout=60.0
            )

            if response.failed > 0:
                logger.warning(
                    f"Import completed with {response.failed} failures: "
                    f"{[e.error for e in response.errors]}"
                )

            return response
        except grpc.RpcError as e:
            logger.error(f"Failed to import products: {e.details()}")
            raise

    async def watch_inventory(
        self,
        updates: AsyncIterator[pb.InventoryUpdate]
    ) -> AsyncIterator[pb.InventoryChange]:
        """Bidirectional streaming for inventory monitoring."""
        try:
            stream = self._get_stub().WatchInventory(
                metadata=self._get_metadata()
            )

            # Start sending updates
            send_task = asyncio.create_task(self._send_updates(stream, updates))

            # Receive changes
            try:
                async for change in stream:
                    yield change
            finally:
                send_task.cancel()

        except grpc.RpcError as e:
            logger.error(f"Failed in inventory watch: {e.details()}")
            raise

    async def _send_updates(
        self,
        stream,
        updates: AsyncIterator[pb.InventoryUpdate]
    ):
        """Send inventory updates to server."""
        try:
            async for update in updates:
                await stream.write(update)
            await stream.done_writing()
        except asyncio.CancelledError:
            pass

    async def close(self):
        """Close all channels gracefully."""
        close_tasks = [channel.close() for channel in self.channels]
        await asyncio.gather(*close_tasks)

    @staticmethod
    def _generate_trace_id() -> str:
        """Generate unique trace ID for request tracking."""
        import uuid
        return str(uuid.uuid4())

# Usage example
async def main():
    # Initialize client with load balancing
    client = ProductClient(
        servers=[
            'product-service-1:50051',
            'product-service-2:50051',
            'product-service-3:50051'
        ],
        api_key='your-api-key',
        use_tls=True
    )

    try:
        # Unary call
        product = await client.get_product('prod-123')
        print(f"Product: {product.name} - ${product.price}")

        # Server streaming
        async for product in client.list_products(
            category='electronics',
            min_price=100,
            max_price=1000
        ):
            print(f"Listed: {product.name}")

        # Client streaming
        products_to_import = [
            pb.Product(id=f'new-{i}', name=f'Product {i}', price=99.99)
            for i in range(100)
        ]
        summary = await client.import_products(products_to_import)
        print(f"Imported {summary.successful} products")

        # Bidirectional streaming
        async def generate_updates():
            for i in range(10):
                yield pb.InventoryUpdate(
                    product_id=f'prod-{i}',
                    quantity_change=5,
                    warehouse_id='warehouse-1'
                )
                await asyncio.sleep(1)

        async for change in client.watch_inventory(generate_updates()):
            print(f"Inventory change: {change.product_id} -> {change.new_quantity}")

    finally:
        await client.close()

if __name__ == '__main__':
    asyncio.run(main())

Example 3: Node.js Implementation with Health Checking

// server/index.js
const grpc = require('@grpc/grpc-js');
const protoLoader = require('@grpc/proto-loader');
const path = require('path');

// Load proto file
const PROTO_PATH = path.join(__dirname, '../proto/ecommerce.proto');
const packageDefinition = protoLoader.loadSync(PROTO_PATH, {
    keepCase: true,
    longs: String,
    enums: String,
    defaults: true,
    oneofs: true
});

const protoDescriptor = grpc.loadPackageDefinition(packageDefinition);
const ecommerce = protoDescriptor.ecommerce.v1;

// Health check implementation
const health = require('grpc-health-check');
const healthImpl = new health.Implementation({
    '': 'SERVING',
    'ecommerce.v1.ProductService': 'SERVING'
});

// Service implementation
class ProductService {
    constructor() {
        this.products = new Map();
        this.watchers = new Map();
    }

    async getProduct(call, callback) {
        const { product_id } = call.request;
        const product = this.products.get(product_id);

        if (!product) {
            callback({
                code: grpc.status.NOT_FOUND,
                message: `Product ${product_id} not found`
            });
            return;
        }

        callback(null, product);
    }

    async listProducts(call) {
        const { category, page_size } = call.request;
        let count = 0;

        for (const [id, product] of this.products) {
            if (category && product.categories.indexOf(category) === -1) {
                continue;
            }

            call.write(product);
            count++;

            if (page_size > 0 && count >= page_size) {
                break;
            }

            // Simulate processing delay
            await new Promise(resolve => setTimeout(resolve, 10));
        }

        call.end();
    }

    // Add remaining methods...
}

// Server setup
function main() {
    const server = new grpc.Server({
        'grpc.max_concurrent_streams': 1000,
        'grpc.max_receive_message_length': 1024 * 1024 * 16
    });

    // Add services
    server.addService(
        ecommerce.ProductService.service,
        new ProductService()
    );

    // Add health check
    server.addService(health.service, healthImpl);

    // Start server
    server.bindAsync(
        '0.0.0.0:50051',
        grpc.ServerCredentials.createInsecure(),
        (err, port) => {
            if (err) {
                console.error('Failed to bind:', err);
                return;
            }
            console.log(`gRPC server running on port ${port}`);
            server.start();
        }
    );
}

main();

Error Handling

Error	Cause	Solution
"Failed to compile proto"	Invalid Protocol Buffer syntax	Validate with protoc --lint
"Connection refused"	Server not running or wrong port	Check server status and port
"Deadline exceeded"	Request timeout	Increase timeout or optimize operation
"Resource exhausted"	Rate limiting or quota exceeded	Implement backoff and retry
"Unavailable"	Server temporarily down	Implement circuit breaker pattern

Configuration Options

Server Options

• MaxConcurrentStreams: Limit concurrent streams per connection
• MaxReceiveMessageSize: Maximum message size (default 4MB)
• KeepaliveParams: Connection health monitoring
• ConnectionTimeout: Maximum idle time before closing

Client Options

• LoadBalancingPolicy: round_robin, pick_first, grpclb
• WaitForReady: Block until server available
• Retry: Automatic retry configuration
• Interceptors: Add cross-cutting concerns

Best Practices

DO:

• Use field numbers consistently for backward compatibility
• Implement proper error codes and messages
• Add request deadlines for all RPCs
• Use streaming for large datasets
• Implement health checking endpoints
• Version your services properly

DON'T:

• Change field numbers in proto files
• Use gRPC for browser clients without proxy
• Ignore proper error handling
• Send large messages without streaming
• Skip TLS in production
• Use synchronous calls for long operations

Performance Considerations

• Binary protocol reduces bandwidth by 20-30% vs JSON
• HTTP/2 multiplexing eliminates head-of-line blocking
• Connection pooling reduces handshake overhead
• Streaming prevents memory exhaustion with large datasets
• Protocol Buffers provide 3-10x faster serialization than JSON

Security Considerations

• Always use TLS in production with mutual authentication
• Implement token-based authentication via metadata
• Use interceptors for consistent auth across services
• Validate all input according to proto definitions
• Implement rate limiting per client
• Use service accounts for service-to-service auth

Related Commands

• /rest-api-generator - Generate REST APIs
• /graphql-server-builder - Build GraphQL servers
• /api-gateway-builder - Create API gateways
• /webhook-handler-creator - Handle webhooks
• /websocket-server-builder - WebSocket servers

Version History

• v1.0.0 (2024-10): Initial implementation with Go, Python, Node.js support
• Planned v1.1.0: Add Rust and Java implementations with advanced load balancing