Grpc Microservices

func (s *server) SearchProducts(req *pb.SearchRequest, stream pb.ProductService_SearchProductsServer) error {    products := s.db.Search(req.Query, req.Limit)
    for _, product := range products {        if err := stream.Send(&product); err != nil {            return err        }    }
    return nil}

func (s *server) UploadImages(stream pb.UploadService_UploadImagesServer) error {    var count int32    var totalBytes int64    var filenames []string
    for {        chunk, err := stream.Recv()        if err == io.EOF {            return stream.SendAndClose(&pb.UploadSummary{                TotalImages: count,                TotalBytes:  totalBytes,                UploadedFilenames: filenames,            })        }        if err != nil {            return err        }
        // Process chunk        totalBytes += int64(len(chunk.Data))        if chunk.ChunkNumber == 0 {            count++            filenames = append(filenames, chunk.Filename)        }    }}

func (s *server) Chat(stream pb.ChatService_ChatServer) error {    // Create channel for this client    clientID := uuid.New().String()    msgChan := make(chan *pb.ChatMessage, 10)
    // Register client    s.mu.Lock()    s.clients[clientID] = msgChan    s.mu.Unlock()
    defer func() {        s.mu.Lock()        delete(s.clients, clientID)        close(msgChan)        s.mu.Unlock()    }()
    // Goroutine to send messages to client    go func() {        for msg := range msgChan {            if err := stream.Send(msg); err != nil {                return            }        }    }()
    // Receive messages from client    for {        msg, err := stream.Recv()        if err == io.EOF {            return nil        }        if err != nil {            return err        }
        // Broadcast to all clients in room        s.broadcast(msg)    }}

func UnaryServerInterceptor() grpc.UnaryServerInterceptor {    return func(        ctx context.Context,        req interface{},        info *grpc.UnaryServerInfo,        handler grpc.UnaryHandler,    ) (interface{}, error) {        // Pre-processing        start := time.Now()        log.Printf("Method: %s, Start: %v", info.FullMethod, start)
        // Call the handler        resp, err := handler(ctx, req)
        // Post-processing        duration := time.Since(start)        log.Printf("Method: %s, Duration: %v, Error: %v",            info.FullMethod, duration, err)
        return resp, err    }}
// Usageserver := grpc.NewServer(    grpc.UnaryInterceptor(UnaryServerInterceptor()),)

func UnaryClientInterceptor() grpc.UnaryClientInterceptor {    return func(        ctx context.Context,        method string,        req, reply interface{},        cc *grpc.ClientConn,        invoker grpc.UnaryInvoker,        opts ...grpc.CallOption,    ) error {        start := time.Now()
        // Call the remote method        err := invoker(ctx, method, req, reply, cc, opts...)
        log.Printf("Method: %s, Duration: %v, Error: %v",            method, time.Since(start), err)
        return err    }}
// Usageconn, err := grpc.Dial(    address,    grpc.WithUnaryInterceptor(UnaryClientInterceptor()),)

func StreamServerInterceptor() grpc.StreamServerInterceptor {    return func(        srv interface{},        ss grpc.ServerStream,        info *grpc.StreamServerInfo,        handler grpc.StreamHandler,    ) error {        log.Printf("Stream started: %s", info.FullMethod)
        err := handler(srv, ss)
        log.Printf("Stream ended: %s, Error: %v", info.FullMethod, err)
        return err    }}

func AuthInterceptor(secret string) grpc.UnaryServerInterceptor {    return func(        ctx context.Context,        req interface{},        info *grpc.UnaryServerInfo,        handler grpc.UnaryHandler,    ) (interface{}, error) {        // Extract metadata        md, ok := metadata.FromIncomingContext(ctx)        if !ok {            return nil, status.Error(codes.Unauthenticated, "no metadata")        }
        // Get authorization token        tokens := md["authorization"]        if len(tokens) == 0 {            return nil, status.Error(codes.Unauthenticated, "no token")        }
        // Validate token        token := tokens[0]        claims, err := validateJWT(token, secret)        if err != nil {            return nil, status.Error(codes.Unauthenticated, "invalid token")        }
        // Add claims to context        ctx = context.WithValue(ctx, "claims", claims)
        return handler(ctx, req)    }}

func LoggingInterceptor(logger *log.Logger) grpc.UnaryServerInterceptor {    return func(        ctx context.Context,        req interface{},        info *grpc.UnaryServerInfo,        handler grpc.UnaryHandler,    ) (interface{}, error) {        start := time.Now()
        // Get request ID from metadata        requestID := getRequestID(ctx)
        logger.Printf("[%s] Request: %s", requestID, info.FullMethod)
        resp, err := handler(ctx, req)
        duration := time.Since(start)        statusCode := status.Code(err)
        logger.Printf("[%s] Response: %s, Duration: %v, Status: %v",            requestID, info.FullMethod, duration, statusCode)
        return resp, err    }}

func RateLimitInterceptor(limiter *rate.Limiter) grpc.UnaryServerInterceptor {    return func(        ctx context.Context,        req interface{},        info *grpc.UnaryServerInfo,        handler grpc.UnaryHandler,    ) (interface{}, error) {        if !limiter.Allow() {            return nil, status.Error(                codes.ResourceExhausted,                "rate limit exceeded",            )        }
        return handler(ctx, req)    }}

func TracingInterceptor(tracer trace.Tracer) grpc.UnaryServerInterceptor {    return func(        ctx context.Context,        req interface{},        info *grpc.UnaryServerInfo,        handler grpc.UnaryHandler,    ) (interface{}, error) {        ctx, span := tracer.Start(ctx, info.FullMethod)        defer span.End()
        // Add attributes        span.SetAttributes(            attribute.String("rpc.method", info.FullMethod),            attribute.String("rpc.service", "MyService"),        )
        resp, err := handler(ctx, req)
        if err != nil {            span.RecordError(err)            span.SetStatus(codes2.Error, err.Error())        } else {            span.SetStatus(codes2.Ok, "")        }
        return resp, err    }}

func RecoveryInterceptor() grpc.UnaryServerInterceptor {    return func(        ctx context.Context,        req interface{},        info *grpc.UnaryServerInfo,        handler grpc.UnaryHandler,    ) (resp interface{}, err error) {        defer func() {            if r := recover(); r != nil {                log.Printf("Panic recovered: %v\n%s", r, debug.Stack())                err = status.Error(codes.Internal, "internal server error")            }        }()
        return handler(ctx, req)    }}

server := grpc.NewServer(    grpc.ChainUnaryInterceptor(        RecoveryInterceptor(),        LoggingInterceptor(logger),        TracingInterceptor(tracer),        AuthInterceptor(jwtSecret),        RateLimitInterceptor(limiter),    ),    grpc.ChainStreamInterceptor(        StreamRecoveryInterceptor(),        StreamLoggingInterceptor(logger),    ),)

import "google.golang.org/grpc/balancer/roundrobin"
conn, err := grpc.Dial(    "dns:///my-service.example.com:50051",    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),    grpc.WithInsecure(),)

conn, err := grpc.Dial(    "dns:///my-service.example.com:50051",    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"pick_first"}`),    grpc.WithInsecure(),)

type exampleResolver struct {    target     resolver.Target    cc         resolver.ClientConn    addrsStore map[string][]string}
func (r *exampleResolver) ResolveNow(resolver.ResolveNowOptions) {    // Discover service addresses    addresses := r.discoverServices()
    var addrs []resolver.Address    for _, addr := range addresses {        addrs = append(addrs, resolver.Address{Addr: addr})    }
    r.cc.UpdateState(resolver.State{Addresses: addrs})}
func init() {    resolver.Register(&exampleResolverBuilder{})}

import "google.golang.org/grpc/health"import healthpb "google.golang.org/grpc/health/grpc_health_v1"
healthServer := health.NewServer()healthpb.RegisterHealthServer(grpcServer, healthServer)
// Set service statushealthServer.SetServingStatus("UserService", healthpb.HealthCheckResponse_SERVING)

import "google.golang.org/grpc/codes"import "google.golang.org/grpc/status"
// Return errors with appropriate codesfunc (s *server) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {    if req.Id <= 0 {        return nil, status.Error(codes.InvalidArgument, "id must be positive")    }
    user, err := s.db.GetUser(req.Id)    if err == sql.ErrNoRows {        return nil, status.Error(codes.NotFound, "user not found")    }    if err != nil {        return nil, status.Error(codes.Internal, "database error")    }
    return user, nil}

import "google.golang.org/genproto/googleapis/rpc/errdetails"
func (s *server) CreateUser(ctx context.Context, req *pb.CreateUserRequest) (*pb.User, error) {    // Validate request    violations := validateCreateUserRequest(req)    if len(violations) > 0 {        badRequest := &errdetails.BadRequest{}        for field, msg := range violations {            badRequest.FieldViolations = append(                badRequest.FieldViolations,                &errdetails.BadRequest_FieldViolation{                    Field:       field,                    Description: msg,                },            )        }
        st := status.New(codes.InvalidArgument, "invalid request")        st, _ = st.WithDetails(badRequest)        return nil, st.Err()    }
    // Create user...}

resp, err := client.CreateUser(ctx, req)if err != nil {    st := status.Convert(err)
    for _, detail := range st.Details() {        switch t := detail.(type) {        case *errdetails.BadRequest:            for _, violation := range t.FieldViolations {                fmt.Printf("Invalid field %s: %s\n",                    violation.Field, violation.Description)            }        }    }}

func (s *server) ProcessOrder(ctx context.Context, req *pb.OrderRequest) (*pb.OrderResponse, error) {    // Call inventory service    inventory, err := s.inventoryClient.CheckInventory(ctx, &pb.InventoryRequest{        ProductId: req.ProductId,    })    if err != nil {        // Propagate error with additional context        st := status.Convert(err)        return nil, status.Errorf(st.Code(),            "inventory check failed: %v", st.Message())    }
    // Continue processing...}

import "google.golang.org/grpc/codes"import "google.golang.org/grpc/status"
func CallWithRetry(ctx context.Context, maxRetries int, fn func() error) error {    var err error
    for i := 0; i < maxRetries; i++ {        err = fn()        if err == nil {            return nil        }
        // Check if error is retryable        st := status.Convert(err)        if !isRetryable(st.Code()) {            return err        }
        // Exponential backoff        backoff := time.Duration(math.Pow(2, float64(i))) * time.Second        time.Sleep(backoff)    }
    return err}
func isRetryable(code codes.Code) bool {    return code == codes.Unavailable ||           code == codes.DeadlineExceeded ||           code == codes.ResourceExhausted}

// Reuse connectionsvar conn *grpc.ClientConnvar once sync.Once
func getConnection() *grpc.ClientConn {    once.Do(func() {        var err error        conn, err = grpc.Dial(            address,            grpc.WithKeepaliveParams(keepalive.ClientParameters{                Time:                10 * time.Second,                Timeout:             3 * time.Second,                PermitWithoutStream: true,            }),        )        if err != nil {            log.Fatal(err)        }    })    return conn}

type ConnectionPool struct {    connections []*grpc.ClientConn    next        uint32}
func (p *ConnectionPool) GetConnection() *grpc.ClientConn {    n := atomic.AddUint32(&p.next, 1)    return p.connections[n%uint32(len(p.connections))]}

// Server-sidecreds, err := credentials.NewServerTLSFromFile(certFile, keyFile)server := grpc.NewServer(grpc.Creds(creds))
// Client-sidecreds, err := credentials.NewClientTLSFromFile(certFile, "")conn, err := grpc.Dial(address, grpc.WithTransportCredentials(creds))

cert, err := tls.LoadX509KeyPair(certFile, keyFile)certPool := x509.NewCertPool()ca, err := ioutil.ReadFile(caFile)certPool.AppendCertsFromPEM(ca)
creds := credentials.NewTLS(&tls.Config{    Certificates: []tls.Certificate{cert},    ClientAuth:   tls.RequireAndVerifyClientCert,    ClientCAs:    certPool,})
server := grpc.NewServer(grpc.Creds(creds))

type tokenAuth struct {    token string}
func (t tokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {    return map[string]string{        "authorization": "Bearer " + t.token,    }, nil}
func (t tokenAuth) RequireTransportSecurity() bool {    return true}
// Usageconn, err := grpc.Dial(    address,    grpc.WithPerRPCCredentials(tokenAuth{token: "my-token"}),)

// Set deadline for requestctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)defer cancel()
resp, err := client.GetUser(ctx, &pb.GetUserRequest{Id: 123})if err != nil {    if status.Code(err) == codes.DeadlineExceeded {        log.Println("Request timed out")    }}

func (s *server) ComplexOperation(ctx context.Context, req *pb.Request) (*pb.Response, error) {    // Check if deadline is already exceeded    deadline, ok := ctx.Deadline()    if ok && time.Now().After(deadline) {        return nil, status.Error(codes.DeadlineExceeded, "deadline exceeded")    }
    // Propagate context to downstream calls    user, err := s.userClient.GetUser(ctx, &pb.GetUserRequest{Id: req.UserId})    if err != nil {        return nil, err    }
    // Continue with remaining time...}

import "github.com/grpc-ecosystem/go-grpc-prometheus"
// Server metricsgrpcMetrics := grpc_prometheus.NewServerMetrics()server := grpc.NewServer(    grpc.UnaryInterceptor(grpcMetrics.UnaryServerInterceptor()),    grpc.StreamInterceptor(grpcMetrics.StreamServerInterceptor()),)grpcMetrics.InitializeMetrics(server)
// Expose metricshttp.Handle("/metrics", promhttp.Handler())

server := grpc.NewServer()// Register services...
go func() {    if err := server.Serve(listener); err != nil {        log.Fatalf("failed to serve: %v", err)    }}()
// Wait for interrupt signalquit := make(chan os.Signal, 1)signal.Notify(quit, os.Interrupt, syscall.SIGTERM)<-quit
log.Println("Shutting down server...")
// Graceful shutdownserver.GracefulStop()log.Println("Server stopped")

type mockUserClient struct {    pb.UserServiceClient    getUserFunc func(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error)}
func (m *mockUserClient) GetUser(ctx context.Context, req *pb.GetUserRequest, opts ...grpc.CallOption) (*pb.User, error) {    return m.getUserFunc(ctx, req)}
func TestOrderService(t *testing.T) {    mockClient := &mockUserClient{        getUserFunc: func(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {            return &pb.User{Id: 1, Name: "Test User"}, nil        },    }
    // Test with mock...}

func TestIntegration(t *testing.T) {    // Start test server    lis, err := net.Listen("tcp", ":0")    require.NoError(t, err)
    server := grpc.NewServer()    pb.RegisterUserServiceServer(server, &userServer{})
    go server.Serve(lis)    defer server.Stop()
    // Connect client    conn, err := grpc.Dial(lis.Addr().String(), grpc.WithInsecure())    require.NoError(t, err)    defer conn.Close()
    client := pb.NewUserServiceClient(conn)
    // Test requests...}