// src/server.ts
import * as grpc from '@grpc/grpc-js';
import { UserServiceService, IUserServiceServer } from './generated/user_service';
import { User, GetUserResponse, ListUsersResponse } from './generated/user';
import { GetUserRequest, CreateUserRequest, ListUsersRequest } from './generated/user_service';

// Mock database
const usersDb = new Map<number, any>([
  [1, { id: 1, name: 'Alice', email: 'alice@example.com', isActive: true }],
  [2, { id: 2, name: 'Bob', email: 'bob@example.com', isActive: true }],
]);

let nextUserId = 3;

// Implement service handlers
const userServiceImpl: IUserServiceServer = {
  // Unary RPC
  async getUser(
    call: grpc.ServerUnaryCall<GetUserRequest, GetUserResponse>,
    callback: grpc.sendUnaryData<GetUserResponse>
  ) {
    const { userId } = call.request;
    const user = usersDb.get(userId);

    if (!user) {
      return callback({
        code: grpc.status.NOT_FOUND,
        message: `User ${userId} not found`,
      });
    }

    callback(null, { user });
  },

  // Unary RPC with validation
  async createUser(
    call: grpc.ServerUnaryCall<CreateUserRequest, GetUserResponse>,
    callback: grpc.sendUnaryData<GetUserResponse>
  ) {
    const { name, email } = call.request;

    if (!name || !email) {
      return callback({
        code: grpc.status.INVALID_ARGUMENT,
        message: 'Name and email are required',
      });
    }

    const user = {
      id: nextUserId++,
      name,
      email,
      isActive: true,
      createdAt: new Date(),
    };

    usersDb.set(user.id, user);
    callback(null, { user });
  },

  // Server streaming
  async listUsers(call: grpc.ServerWritableStream<ListUsersRequest, GetUserResponse>) {
    const { page = 1, pageSize = 10 } = call.request;
    const users = Array.from(usersDb.values());
    const start = (page - 1) * pageSize;
    const paginatedUsers = users.slice(start, start + pageSize);

    for (const user of paginatedUsers) {
      call.write({ user });
    }

    call.end();
  },

  // Client streaming
  async batchCreateUsers(
    call: grpc.ServerReadableStream<CreateUserRequest, ListUsersResponse>,
    callback: grpc.sendUnaryData<ListUsersResponse>
  ) {
    const users: any[] = [];

    call.on('data', (request: CreateUserRequest) => {
      const user = {
        id: nextUserId++,
        name: request.name,
        email: request.email,
        isActive: true,
      };
      users.push(user);
      usersDb.set(user.id, user);
    });

    call.on('end', () => {
      callback(null, { users, totalCount: users.length });
    });

    call.on('error', (err) => {
      console.error('Stream error:', err);
      callback({
        code: grpc.status.INTERNAL,
        message: 'Stream error',
      });
    });
  },

  // Bidirectional streaming
  async syncUsers(call: grpc.ServerDuplexStream<any, any>) {
    call.on('data', (request: any) => {
      const user = {
        id: request.userId || nextUserId++,
        name: request.name,
        email: request.email,
        isActive: true,
      };

      usersDb.set(user.id, user);
      call.write({
        userId: user.id,
        status: 'SYNCED',
        message: `User ${user.name} synced successfully`,
      });
    });

    call.on('end', () => {
      call.end();
    });

    call.on('error', (err) => {
      console.error('Sync error:', err);
    });
  },

  async deleteUser(
    call: grpc.ServerUnaryCall<any, any>,
    callback: grpc.sendUnaryData<any>
  ) {
    const { userId } = call.request;

    if (usersDb.has(userId)) {
      usersDb.delete(userId);
      return callback(null, {});
    }

    callback({
      code: grpc.status.NOT_FOUND,
      message: `User ${userId} not found`,
    });
  },
};

// Create and start server
function startServer() {
  const server = new grpc.Server();

  server.addService(UserServiceService, userServiceImpl);

  const PORT = process.env.PORT || 50051;
  server.bindAsync(`0.0.0.0:${PORT}`, grpc.ServerCredentials.createInsecure(), (err) => {
    if (err) {
      console.error('Failed to bind server:', err);
      process.exit(1);
    }

    server.start();
    console.log(`gRPC server running on port ${PORT}`);
  });
}

startServer();

// src/main/java/com/example/grpc/UserServiceServer.java
package com.example.grpc;

import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import users.UserServiceGrpc;
import users.UserServiceProto.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

public class UserServiceServer {

    private static final Map<Integer, User> usersDb = new ConcurrentHashMap<>();
    private static final AtomicInteger nextUserId = new AtomicInteger(3);

    static {
        usersDb.put(1, User.newBuilder().setId(1).setName("Alice").setEmail("alice@example.com").setIsActive(true).build());
        usersDb.put(2, User.newBuilder().setId(2).setName("Bob").setEmail("bob@example.com").setIsActive(true).build());
    }

    static class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {

        @Override
        public void getUser(GetUserRequest request, StreamObserver<GetUserResponse> responseObserver) {
            int userId = request.getUserId();
            User user = usersDb.get(userId);

            if (user == null) {
                responseObserver.onError(Status.NOT_FOUND
                    .withDescription("User " + userId + " not found")
                    .asRuntimeException());
                return;
            }

            responseObserver.onNext(GetUserResponse.newBuilder().setUser(user).build());
            responseObserver.onCompleted();
        }

        @Override
        public void createUser(CreateUserRequest request, StreamObserver<GetUserResponse> responseObserver) {
            if (request.getName().isEmpty() || request.getEmail().isEmpty()) {
                responseObserver.onError(Status.INVALID_ARGUMENT
                    .withDescription("Name and email are required")
                    .asRuntimeException());
                return;
            }

            int id = nextUserId.getAndIncrement();
            User user = User.newBuilder()
                .setId(id)
                .setName(request.getName())
                .setEmail(request.getEmail())
                .setIsActive(true)
                .build();

            usersDb.put(id, user);
            responseObserver.onNext(GetUserResponse.newBuilder().setUser(user).build());
            responseObserver.onCompleted();
        }

        @Override
        public void listUsers(ListUsersRequest request, StreamObserver<GetUserResponse> responseObserver) {
            int page = request.getPage() > 0 ? request.getPage() : 1;
            int pageSize = request.getPageSize() > 0 ? request.getPageSize() : 10;
            List<User> users = new ArrayList<>(usersDb.values());
            int start = (page - 1) * pageSize;
            int end = Math.min(start + pageSize, users.size());

            for (int i = start; i < end; i++) {
                responseObserver.onNext(GetUserResponse.newBuilder().setUser(users.get(i)).build());
            }
            responseObserver.onCompleted();
        }

        @Override
        public StreamObserver<CreateUserRequest> batchCreateUsers(StreamObserver<ListUsersResponse> responseObserver) {
            List<User> createdUsers = new ArrayList<>();

            return new StreamObserver<>() {
                @Override
                public void onNext(CreateUserRequest request) {
                    int id = nextUserId.getAndIncrement();
                    User user = User.newBuilder()
                        .setId(id)
                        .setName(request.getName())
                        .setEmail(request.getEmail())
                        .setIsActive(true)
                        .build();
                    createdUsers.add(user);
                    usersDb.put(id, user);
                }

                @Override
                public void onError(Throwable t) {
                    responseObserver.onError(Status.INTERNAL.withDescription("Stream error").asRuntimeException());
                }

                @Override
                public void onCompleted() {
                    responseObserver.onNext(ListUsersResponse.newBuilder()
                        .addAllUsers(createdUsers)
                        .setTotalCount(createdUsers.size())
                        .build());
                    responseObserver.onCompleted();
                }
            };
        }

        @Override
        public StreamObserver<SyncUserRequest> syncUsers(StreamObserver<SyncUserResponse> responseObserver) {
            return new StreamObserver<>() {
                @Override
                public void onNext(SyncUserRequest request) {
                    int id = request.getUserId() > 0 ? request.getUserId() : nextUserId.getAndIncrement();
                    User user = User.newBuilder()
                        .setId(id).setName(request.getName()).setEmail(request.getEmail()).setIsActive(true)
                        .build();
                    usersDb.put(id, user);
                    responseObserver.onNext(SyncUserResponse.newBuilder()
                        .setUserId(id).setStatus("SYNCED")
                        .setMessage("User " + user.getName() + " synced successfully")
                        .build());
                }

                @Override public void onError(Throwable t) { System.err.println("Sync error: " + t.getMessage()); }
                @Override public void onCompleted() { responseObserver.onCompleted(); }
            };
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        int port = Integer.parseInt(System.getenv().getOrDefault("PORT", "50051"));
        Server server = ServerBuilder.forPort(port)
            .addService(new UserServiceImpl())
            .build()
            .start();

        System.out.println("gRPC server running on port " + port);
        server.awaitTermination();
    }
}

# src/server.py
import grpc
import users_pb2
import users_pb2_grpc
from concurrent import futures
import os

# Mock database
users_db: dict[int, dict] = {
    1: {"id": 1, "name": "Alice", "email": "alice@example.com", "is_active": True},
    2: {"id": 2, "name": "Bob", "email": "bob@example.com", "is_active": True},
}
next_user_id = 3


class UserServiceServicer(users_pb2_grpc.UserServiceServicer):

    def GetUser(self, request, context):
        user_data = users_db.get(request.user_id)
        if not user_data:
            context.abort(grpc.StatusCode.NOT_FOUND, f"User {request.user_id} not found")
        user = users_pb2.User(**user_data)
        return users_pb2.GetUserResponse(user=user)

    def CreateUser(self, request, context):
        global next_user_id
        if not request.name or not request.email:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Name and email are required")
        user_data = {"id": next_user_id, "name": request.name, "email": request.email, "is_active": True}
        users_db[next_user_id] = user_data
        next_user_id += 1
        return users_pb2.GetUserResponse(user=users_pb2.User(**user_data))

    def ListUsers(self, request, context):
        page = request.page or 1
        page_size = request.page_size or 10
        users = list(users_db.values())
        start = (page - 1) * page_size
        for u in users[start:start + page_size]:
            yield users_pb2.GetUserResponse(user=users_pb2.User(**u))

    def BatchCreateUsers(self, request_iterator, context):
        global next_user_id
        created = []
        for req in request_iterator:
            user_data = {"id": next_user_id, "name": req.name, "email": req.email, "is_active": True}
            users_db[next_user_id] = user_data
            next_user_id += 1
            created.append(users_pb2.User(**user_data))
        return users_pb2.ListUsersResponse(users=created, total_count=len(created))

    def SyncUsers(self, request_iterator, context):
        global next_user_id
        for req in request_iterator:
            uid = req.user_id or next_user_id
            user_data = {"id": uid, "name": req.name, "email": req.email, "is_active": True}
            users_db[uid] = user_data
            if uid == next_user_id:
                next_user_id += 1
            yield users_pb2.SyncUserResponse(
                user_id=uid, status="SYNCED",
                message=f"User {req.name} synced successfully"
            )

    def DeleteUser(self, request, context):
        if request.user_id in users_db:
            del users_db[request.user_id]
            return users_pb2.Empty()
        context.abort(grpc.StatusCode.NOT_FOUND, f"User {request.user_id} not found")


def serve():
    port = os.getenv("PORT", "50051")
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    users_pb2_grpc.add_UserServiceServicer_to_server(UserServiceServicer(), server)
    server.add_insecure_port(f"0.0.0.0:{port}")
    server.start()
    print(f"gRPC server running on port {port}")
    server.wait_for_termination()


if __name__ == "__main__":
    serve()

// src/Services/UserService.cs
using Grpc.Core;
using Users;

namespace GrpcServer.Services;

public class UserServiceImpl : UserService.UserServiceBase
{
    private static readonly Dictionary<int, User> UsersDb = new()
    {
        [1] = new User { Id = 1, Name = "Alice", Email = "alice@example.com", IsActive = true },
        [2] = new User { Id = 2, Name = "Bob", Email = "bob@example.com", IsActive = true },
    };
    private static int _nextUserId = 3;

    public override Task<GetUserResponse> GetUser(GetUserRequest request, ServerCallContext context)
    {
        if (!UsersDb.TryGetValue(request.UserId, out var user))
            throw new RpcException(new Status(StatusCode.NotFound, $"User {request.UserId} not found"));

        return Task.FromResult(new GetUserResponse { User = user });
    }

    public override Task<GetUserResponse> CreateUser(CreateUserRequest request, ServerCallContext context)
    {
        if (string.IsNullOrEmpty(request.Name) || string.IsNullOrEmpty(request.Email))
            throw new RpcException(new Status(StatusCode.InvalidArgument, "Name and email are required"));

        var user = new User { Id = _nextUserId++, Name = request.Name, Email = request.Email, IsActive = true };
        UsersDb[user.Id] = user;
        return Task.FromResult(new GetUserResponse { User = user });
    }

    public override async Task ListUsers(ListUsersRequest request, IServerStreamWriter<GetUserResponse> responseStream, ServerCallContext context)
    {
        int page = request.Page > 0 ? request.Page : 1;
        int pageSize = request.PageSize > 0 ? request.PageSize : 10;
        var users = UsersDb.Values.Skip((page - 1) * pageSize).Take(pageSize);

        foreach (var user in users)
            await responseStream.WriteAsync(new GetUserResponse { User = user });
    }

    public override async Task<ListUsersResponse> BatchCreateUsers(
        IAsyncStreamReader<CreateUserRequest> requestStream, ServerCallContext context)
    {
        var created = new List<User>();
        await foreach (var req in requestStream.ReadAllAsync())
        {
            var user = new User { Id = _nextUserId++, Name = req.Name, Email = req.Email, IsActive = true };
            UsersDb[user.Id] = user;
            created.Add(user);
        }
        var response = new ListUsersResponse { TotalCount = created.Count };
        response.Users.AddRange(created);
        return response;
    }

    public override async Task SyncUsers(
        IAsyncStreamReader<SyncUserRequest> requestStream,
        IServerStreamWriter<SyncUserResponse> responseStream,
        ServerCallContext context)
    {
        await foreach (var req in requestStream.ReadAllAsync())
        {
            int uid = req.UserId > 0 ? req.UserId : _nextUserId++;
            UsersDb[uid] = new User { Id = uid, Name = req.Name, Email = req.Email, IsActive = true };
            await responseStream.WriteAsync(new SyncUserResponse
            {
                UserId = uid, Status = "SYNCED",
                Message = $"User {req.Name} synced successfully"
            });
        }
    }

    public override Task<Empty> DeleteUser(DeleteUserRequest request, ServerCallContext context)
    {
        if (!UsersDb.Remove(request.UserId))
            throw new RpcException(new Status(StatusCode.NotFound, $"User {request.UserId} not found"));

        return Task.FromResult(new Empty());
    }
}

// Program.cs
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddGrpc();
var app = builder.Build();
app.MapGrpcService<UserServiceImpl>();
int port = int.Parse(Environment.GetEnvironmentVariable("PORT") ?? "50051");
app.Run($"http://0.0.0.0:{port}");

// src/client.ts
import * as grpc from '@grpc/grpc-js';
import { UserServiceClient } from './generated/user_service';

class UserServiceClient {
  private client: UserServiceClient;

  constructor(address: string = 'localhost:50051') {
    this.client = new UserServiceClient(
      address,
      grpc.credentials.createInsecure()
    );
  }

  // Unary call
  async getUser(userId: number): Promise<any> {
    return new Promise((resolve, reject) => {
      this.client.getUser({ userId }, (err, response) => {
        if (err) reject(err);
        else resolve(response?.user);
      });
    });
  }

  // Server streaming
  async listUsers(page: number = 1, pageSize: number = 10): Promise<any[]> {
    return new Promise((resolve, reject) => {
      const users: any[] = [];
      const call = this.client.listUsers({ page, pageSize });

      call.on('data', (response) => {
        users.push(response.user);
      });

      call.on('end', () => {
        resolve(users);
      });

      call.on('error', reject);
    });
  }

  // Client streaming
  async batchCreateUsers(requests: any[]): Promise<any> {
    return new Promise((resolve, reject) => {
      const call = this.client.batchCreateUsers((err, response) => {
        if (err) reject(err);
        else resolve(response);
      });

      for (const request of requests) {
        call.write(request);
      }

      call.end();
    });
  }

  // Bidirectional streaming
  async syncUsers(userUpdates: any[]): Promise<void> {
    return new Promise((resolve, reject) => {
      const call = this.client.syncUsers();

      call.on('data', (response) => {
        console.log('Sync response:', response);
      });

      call.on('end', resolve);
      call.on('error', reject);

      for (const update of userUpdates) {
        call.write(update);
      }

      call.end();
    });
  }

  close() {
    grpc.closeClient(this.client);
  }
}

// Usage
async function main() {
  const client = new UserServiceClient();

  try {
    const user = await client.getUser(1);
    console.log('User:', user);

    const users = await client.listUsers(1, 5);
    console.log('Users:', users);
  } catch (error) {
    console.error('Error:', error);
  } finally {
    client.close();
  }
}

main();

// src/main/java/com/example/grpc/UserServiceClientExample.java
package com.example.grpc;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import users.UserServiceGrpc;
import users.UserServiceProto.*;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class UserServiceClientExample {

    private final ManagedChannel channel;
    private final UserServiceGrpc.UserServiceBlockingStub blockingStub;
    private final UserServiceGrpc.UserServiceStub asyncStub;

    public UserServiceClientExample(String address) {
        channel = ManagedChannelBuilder.forTarget(address).usePlaintext().build();
        blockingStub = UserServiceGrpc.newBlockingStub(channel);
        asyncStub = UserServiceGrpc.newStub(channel);
    }

    // Unary call
    public User getUser(int userId) {
        GetUserResponse response = blockingStub.getUser(
            GetUserRequest.newBuilder().setUserId(userId).build()
        );
        return response.getUser();
    }

    // Server streaming
    public List<User> listUsers(int page, int pageSize) {
        List<User> users = new ArrayList<>();
        blockingStub.listUsers(ListUsersRequest.newBuilder().setPage(page).setPageSize(pageSize).build())
            .forEachRemaining(r -> users.add(r.getUser()));
        return users;
    }

    // Client streaming
    public ListUsersResponse batchCreateUsers(List<CreateUserRequest> requests) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        List<ListUsersResponse> result = new ArrayList<>();

        StreamObserver<CreateUserRequest> requestObserver = asyncStub.batchCreateUsers(
            new StreamObserver<>() {
                @Override public void onNext(ListUsersResponse r) { result.add(r); }
                @Override public void onError(Throwable t) { latch.countDown(); }
                @Override public void onCompleted() { latch.countDown(); }
            }
        );

        for (CreateUserRequest req : requests) requestObserver.onNext(req);
        requestObserver.onCompleted();
        latch.await(30, TimeUnit.SECONDS);
        return result.isEmpty() ? null : result.get(0);
    }

    // Bidirectional streaming
    public void syncUsers(List<SyncUserRequest> updates) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);

        StreamObserver<SyncUserRequest> requestObserver = asyncStub.syncUsers(
            new StreamObserver<>() {
                @Override public void onNext(SyncUserResponse r) { System.out.println("Sync response: " + r); }
                @Override public void onError(Throwable t) { latch.countDown(); }
                @Override public void onCompleted() { latch.countDown(); }
            }
        );

        for (SyncUserRequest update : updates) requestObserver.onNext(update);
        requestObserver.onCompleted();
        latch.await(30, TimeUnit.SECONDS);
    }

    public void close() throws InterruptedException {
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }

    public static void main(String[] args) throws InterruptedException {
        UserServiceClientExample client = new UserServiceClientExample("localhost:50051");
        try {
            User user = client.getUser(1);
            System.out.println("User: " + user);

            List<User> users = client.listUsers(1, 5);
            System.out.println("Users: " + users);
        } finally {
            client.close();
        }
    }
}

# src/client.py
import grpc
import users_pb2
import users_pb2_grpc
from typing import Iterator


class UserServiceClient:
    def __init__(self, address: str = "localhost:50051"):
        self.channel = grpc.insecure_channel(address)
        self.stub = users_pb2_grpc.UserServiceStub(self.channel)

    def get_user(self, user_id: int):
        response = self.stub.GetUser(users_pb2.GetUserRequest(user_id=user_id))
        return response.user

    def list_users(self, page: int = 1, page_size: int = 10) -> list:
        users = []
        for response in self.stub.ListUsers(
            users_pb2.ListUsersRequest(page=page, page_size=page_size)
        ):
            users.append(response.user)
        return users

    def batch_create_users(self, requests: list[dict]):
        def request_iterator() -> Iterator[users_pb2.CreateUserRequest]:
            for r in requests:
                yield users_pb2.CreateUserRequest(name=r["name"], email=r["email"])

        return self.stub.BatchCreateUsers(request_iterator())

    def sync_users(self, updates: list[dict]) -> None:
        def request_iterator() -> Iterator[users_pb2.SyncUserRequest]:
            for u in updates:
                yield users_pb2.SyncUserRequest(
                    user_id=u.get("user_id", 0), name=u["name"], email=u["email"]
                )

        for response in self.stub.SyncUsers(request_iterator()):
            print("Sync response:", response)

    def close(self):
        self.channel.close()


def main():
    client = UserServiceClient()
    try:
        user = client.get_user(1)
        print("User:", user)

        users = client.list_users(1, 5)
        print("Users:", users)
    except grpc.RpcError as e:
        print("Error:", e)
    finally:
        client.close()


if __name__ == "__main__":
    main()

// src/Clients/UserServiceClient.cs
using Grpc.Net.Client;
using Users;

namespace GrpcClient;

public class UserServiceClientExample : IDisposable
{
    private readonly GrpcChannel _channel;
    private readonly UserService.UserServiceClient _client;

    public UserServiceClientExample(string address = "http://localhost:50051")
    {
        _channel = GrpcChannel.ForAddress(address);
        _client = new UserService.UserServiceClient(_channel);
    }

    // Unary call
    public async Task<User> GetUserAsync(int userId)
    {
        var response = await _client.GetUserAsync(new GetUserRequest { UserId = userId });
        return response.User;
    }

    // Server streaming
    public async Task<List<User>> ListUsersAsync(int page = 1, int pageSize = 10)
    {
        var users = new List<User>();
        using var call = _client.ListUsers(new ListUsersRequest { Page = page, PageSize = pageSize });
        await foreach (var response in call.ResponseStream.ReadAllAsync())
            users.Add(response.User);
        return users;
    }

    // Client streaming
    public async Task<ListUsersResponse> BatchCreateUsersAsync(IEnumerable<(string Name, string Email)> requests)
    {
        using var call = _client.BatchCreateUsers();
        foreach (var (name, email) in requests)
            await call.RequestStream.WriteAsync(new CreateUserRequest { Name = name, Email = email });
        await call.RequestStream.CompleteAsync();
        return await call.ResponseAsync;
    }

    // Bidirectional streaming
    public async Task SyncUsersAsync(IEnumerable<(int UserId, string Name, string Email)> updates)
    {
        using var call = _client.SyncUsers();
        var readTask = Task.Run(async () =>
        {
            await foreach (var response in call.ResponseStream.ReadAllAsync())
                Console.WriteLine($"Sync response: {response}");
        });

        foreach (var (userId, name, email) in updates)
            await call.RequestStream.WriteAsync(new SyncUserRequest { UserId = userId, Name = name, Email = email });

        await call.RequestStream.CompleteAsync();
        await readTask;
    }

    public void Dispose() => _channel.Dispose();
}

// Program.cs usage
using var client = new UserServiceClientExample();

var user = await client.GetUserAsync(1);
Console.WriteLine($"User: {user}");

var users = await client.ListUsersAsync(1, 5);
Console.WriteLine($"Users: {users.Count}");

// Handler
async getUser(call, callback) {
  const user = await fetchUser(call.request.userId);
  callback(null, { user });
}

// Client
const user = await client.getUser({ userId: 1 });

// Handler
@Override
public void getUser(GetUserRequest request, StreamObserver<GetUserResponse> responseObserver) {
    User user = fetchUser(request.getUserId());
    responseObserver.onNext(GetUserResponse.newBuilder().setUser(user).build());
    responseObserver.onCompleted();
}

// Client (blocking stub)
GetUserResponse response = blockingStub.getUser(
    GetUserRequest.newBuilder().setUserId(1).build()
);
User user = response.getUser();

# Handler
def GetUser(self, request, context):
    user = fetch_user(request.user_id)
    return GetUserResponse(user=user)

# Client
response = stub.GetUser(GetUserRequest(user_id=1))
user = response.user

// Handler
public override Task<GetUserResponse> GetUser(GetUserRequest request, ServerCallContext context)
{
    var user = FetchUser(request.UserId);
    return Task.FromResult(new GetUserResponse { User = user });
}

// Client
var response = await client.GetUserAsync(new GetUserRequest { UserId = 1 });
var user = response.User;

// Handler
async listUsers(call) {
  const users = await fetchAllUsers();
  
  for (const user of users) {
    call.write({ user });
  }
  
  call.end();
}

// Client
const call = client.listUsers({ pageSize: 10 });

call.on('data', (response) => {
  console.log('Received user:', response.user);
});

call.on('end', () => {
  console.log('Stream complete');
});

// Handler
@Override
public void listUsers(ListUsersRequest request, StreamObserver<GetUserResponse> responseObserver) {
    List<User> users = fetchAllUsers();
    for (User user : users) {
        responseObserver.onNext(GetUserResponse.newBuilder().setUser(user).build());
    }
    responseObserver.onCompleted();
}

// Client (blocking iterator)
Iterator<GetUserResponse> responses = blockingStub.listUsers(
    ListUsersRequest.newBuilder().setPageSize(10).build()
);
while (responses.hasNext()) {
    System.out.println("Received user: " + responses.next().getUser());
}
System.out.println("Stream complete");

# Handler
def ListUsers(self, request, context):
    users = fetch_all_users()
    for user in users:
        yield GetUserResponse(user=user)

# Client
for response in stub.ListUsers(ListUsersRequest(page_size=10)):
    print("Received user:", response.user)
print("Stream complete")

// Handler
public override async Task ListUsers(
    ListUsersRequest request,
    IServerStreamWriter<GetUserResponse> responseStream,
    ServerCallContext context)
{
    var users = await FetchAllUsersAsync();
    foreach (var user in users)
        await responseStream.WriteAsync(new GetUserResponse { User = user });
}

// Client
using var call = client.ListUsers(new ListUsersRequest { PageSize = 10 });
await foreach (var response in call.ResponseStream.ReadAllAsync())
    Console.WriteLine($"Received user: {response.User}");
Console.WriteLine("Stream complete");

// Handler
async batchCreateUsers(call, callback) {
  const users = [];
  
  call.on('data', (request) => {
    users.push(request);
  });
  
  call.on('end', () => {
    const saved = saveUsers(users);
    callback(null, { users: saved });
  });
}

// Client
const call = client.batchCreateUsers((err, response) => {
  console.log('All users created:', response.users);
});

call.write({ name: 'Alice', email: 'alice@example.com' });
call.write({ name: 'Bob', email: 'bob@example.com' });
call.end();

// Handler
@Override
public StreamObserver<CreateUserRequest> batchCreateUsers(StreamObserver<ListUsersResponse> responseObserver) {
    List<User> users = new ArrayList<>();
    return new StreamObserver<>() {
        @Override public void onNext(CreateUserRequest req) { users.add(saveUser(req)); }
        @Override public void onError(Throwable t) { responseObserver.onError(t); }
        @Override public void onCompleted() {
            ListUsersResponse resp = ListUsersResponse.newBuilder()
                .addAllUsers(users).setTotalCount(users.size()).build();
            responseObserver.onNext(resp);
            responseObserver.onCompleted();
        }
    };
}

// Client (async stub)
CountDownLatch latch = new CountDownLatch(1);
StreamObserver<CreateUserRequest> req = asyncStub.batchCreateUsers(new StreamObserver<>() {
    @Override public void onNext(ListUsersResponse r) { System.out.println("All users created: " + r.getUsersList()); }
    @Override public void onError(Throwable t) { latch.countDown(); }
    @Override public void onCompleted() { latch.countDown(); }
});
req.onNext(CreateUserRequest.newBuilder().setName("Alice").setEmail("alice@example.com").build());
req.onNext(CreateUserRequest.newBuilder().setName("Bob").setEmail("bob@example.com").build());
req.onCompleted();
latch.await();

# Handler
def BatchCreateUsers(self, request_iterator, context):
    users = []
    for request in request_iterator:
        users.append(save_user(request))
    return ListUsersResponse(users=users, total_count=len(users))

# Client
def request_iterator():
    yield CreateUserRequest(name="Alice", email="alice@example.com")
    yield CreateUserRequest(name="Bob", email="bob@example.com")

response = stub.BatchCreateUsers(request_iterator())
print("All users created:", list(response.users))

// Handler
public override async Task<ListUsersResponse> BatchCreateUsers(
    IAsyncStreamReader<CreateUserRequest> requestStream, ServerCallContext context)
{
    var users = new List<User>();
    await foreach (var req in requestStream.ReadAllAsync())
        users.Add(SaveUser(req));

    var response = new ListUsersResponse { TotalCount = users.Count };
    response.Users.AddRange(users);
    return response;
}

// Client
using var call = client.BatchCreateUsers();
await call.RequestStream.WriteAsync(new CreateUserRequest { Name = "Alice", Email = "alice@example.com" });
await call.RequestStream.WriteAsync(new CreateUserRequest { Name = "Bob", Email = "bob@example.com" });
await call.RequestStream.CompleteAsync();
var response = await call.ResponseAsync;
Console.WriteLine($"All users created: {response.Users.Count}");

// Handler
async syncUsers(call) {
  call.on('data', (request) => {
    const synced = syncUser(request);
    call.write({
      userId: synced.id,
      status: 'SYNCED',
    });
  });
  
  call.on('end', () => {
    call.end();
  });
}

// Client
const call = client.syncUsers();

call.on('data', (response) => {
  console.log('Sync result:', response);
});

call.write({ userId: 1, name: 'Alice Updated' });
call.write({ userId: 2, name: 'Bob Updated' });
call.end();

// Handler
@Override
public StreamObserver<SyncUserRequest> syncUsers(StreamObserver<SyncUserResponse> responseObserver) {
    return new StreamObserver<>() {
        @Override
        public void onNext(SyncUserRequest req) {
            User synced = syncUser(req);
            responseObserver.onNext(SyncUserResponse.newBuilder()
                .setUserId(synced.getId()).setStatus("SYNCED").build());
        }
        @Override public void onError(Throwable t) { responseObserver.onError(t); }
        @Override public void onCompleted() { responseObserver.onCompleted(); }
    };
}

// Client (async stub)
CountDownLatch latch = new CountDownLatch(1);
StreamObserver<SyncUserRequest> req = asyncStub.syncUsers(new StreamObserver<>() {
    @Override public void onNext(SyncUserResponse r) { System.out.println("Sync result: " + r); }
    @Override public void onError(Throwable t) { latch.countDown(); }
    @Override public void onCompleted() { latch.countDown(); }
});
req.onNext(SyncUserRequest.newBuilder().setUserId(1).setName("Alice Updated").build());
req.onNext(SyncUserRequest.newBuilder().setUserId(2).setName("Bob Updated").build());
req.onCompleted();
latch.await();

# Handler
def SyncUsers(self, request_iterator, context):
    for request in request_iterator:
        synced = sync_user(request)
        yield SyncUserResponse(user_id=synced.id, status="SYNCED")

# Client
def request_iterator():
    yield SyncUserRequest(user_id=1, name="Alice Updated")
    yield SyncUserRequest(user_id=2, name="Bob Updated")

for response in stub.SyncUsers(request_iterator()):
    print("Sync result:", response)

// Handler
public override async Task SyncUsers(
    IAsyncStreamReader<SyncUserRequest> requestStream,
    IServerStreamWriter<SyncUserResponse> responseStream,
    ServerCallContext context)
{
    await foreach (var req in requestStream.ReadAllAsync())
    {
        var synced = SyncUser(req);
        await responseStream.WriteAsync(new SyncUserResponse { UserId = synced.Id, Status = "SYNCED" });
    }
}

// Client
using var call = client.SyncUsers();
var readTask = Task.Run(async () =>
{
    await foreach (var response in call.ResponseStream.ReadAllAsync())
        Console.WriteLine($"Sync result: {response}");
});

await call.RequestStream.WriteAsync(new SyncUserRequest { UserId = 1, Name = "Alice Updated" });
await call.RequestStream.WriteAsync(new SyncUserRequest { UserId = 2, Name = "Bob Updated" });
await call.RequestStream.CompleteAsync();
await readTask;

// src/middleware/serverInterceptor.ts
import * as grpc from '@grpc/grpc-js';

interface CallMetadata {
  userId?: string;
  requestId?: string;
  startTime: number;
}

export function createServerInterceptor() {
  return (
    methodDescriptor: any,
    call: any
  ): grpc.Listener => {
    const metadata: CallMetadata = {
      requestId: call.metadata.get('x-request-id')?.[0] as string,
      userId: call.metadata.get('x-user-id')?.[0] as string,
      startTime: Date.now(),
    };

    // Log incoming call
    console.log('[gRPC Server]', {
      method: methodDescriptor.path,
      requestId: metadata.requestId,
      userId: metadata.userId,
    });

    // Wrap call handlers
    const originalSendMetadata = call.sendMetadata;
    call.sendMetadata = function(responseMetadata: grpc.Metadata) {
      responseMetadata.add('x-request-id', metadata.requestId || '');
      originalSendMetadata.call(this, responseMetadata);
    };

    return {
      onReceiveMetadata(metadata: grpc.Metadata, next: any) {
        next(metadata);
      },
      onReceiveMessage(message: any, next: any) {
        console.log('[gRPC Message]', {
          method: methodDescriptor.path,
          requestId: metadata.requestId,
          payload: JSON.stringify(message).substring(0, 200),
        });
        next(message);
      },
      onReceiveHalfClose(next: any) {
        next();
      },
      onCancel(next: any) {
        const duration = Date.now() - metadata.startTime;
        console.log('[gRPC Cancelled]', {
          method: methodDescriptor.path,
          requestId: metadata.requestId,
          duration,
        });
        next();
      },
    };
  };
}

// src/main/java/com/example/grpc/ServerLoggingInterceptor.java
package com.example.grpc;

import io.grpc.*;

public class ServerLoggingInterceptor implements ServerInterceptor {

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call,
            Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {

        String requestId = headers.get(Metadata.Key.of("x-request-id", Metadata.ASCII_STRING_MARSHALLER));
        String userId = headers.get(Metadata.Key.of("x-user-id", Metadata.ASCII_STRING_MARSHALLER));
        long startTime = System.currentTimeMillis();

        System.out.printf("[gRPC Server] method=%s requestId=%s userId=%s%n",
            call.getMethodDescriptor().getFullMethodName(), requestId, userId);

        ServerCall<ReqT, RespT> wrappedCall = new ForwardingServerCall.SimpleForwardingServerCall<>(call) {
            @Override
            public void sendHeaders(Metadata responseHeaders) {
                if (requestId != null) {
                    responseHeaders.put(
                        Metadata.Key.of("x-request-id", Metadata.ASCII_STRING_MARSHALLER), requestId);
                }
                super.sendHeaders(responseHeaders);
            }
        };

        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<>(next.startCall(wrappedCall, headers)) {
            @Override
            public void onMessage(ReqT message) {
                System.out.printf("[gRPC Message] method=%s requestId=%s payload=%s%n",
                    call.getMethodDescriptor().getFullMethodName(), requestId,
                    message.toString().substring(0, Math.min(200, message.toString().length())));
                super.onMessage(message);
            }

            @Override
            public void onCancel() {
                long duration = System.currentTimeMillis() - startTime;
                System.out.printf("[gRPC Cancelled] method=%s requestId=%s duration=%dms%n",
                    call.getMethodDescriptor().getFullMethodName(), requestId, duration);
                super.onCancel();
            }
        };
    }
}