Типы вызовов

service UserService {
    rpc GetUser (GetUserRequest) returns (User);
}

// Клиент (Go)
req := &pb.GetUserRequest{UserId: 123}
resp, err := client.GetUser(ctx, req)
fmt.Println(resp.Name) // "Иван"

service LogService {
    rpc StreamLogs (StreamLogsRequest) returns (stream LogEntry);
}

// Клиент (Go)
stream, err := client.StreamLogs(ctx, &pb.StreamLogsRequest{Level: "ERROR"})
for {
    log, err := stream.Recv()
    if err == io.EOF {
        break // Конец потока
    }
    fmt.Println(log.Message)
}

// Сервер (Go)
func (s *logServer) StreamLogs(req *pb.StreamLogsRequest, stream pb.LogService_StreamLogsServer) error {
    for log := range s.logsChannel {
        if err := stream.Send(log); err != nil {
            return err
        }
    }
    return nil
}

service UploadService {
    rpc UploadFile (stream Chunk) returns (UploadResponse);
}

// Клиент (Go)
stream, err := client.UploadFile(ctx)
for _, chunk := range fileChunks {
    stream.Send(&pb.Chunk{Data: chunk})
}
resp, err := stream.CloseAndRecv()
fmt.Printf("Загружено %d байт", resp.Size)

// Сервер (Go)
func (s *uploadServer) UploadFile(stream pb.UploadService_UploadFileServer) error {
    var totalSize int64
    for {
        chunk, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.UploadResponse{Size: totalSize})
        }
        totalSize += int64(len(chunk.Data))
        // сохраняем chunk в файл
    }
}

service ChatService {
    rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}

// Клиент (Go)
stream, err := client.Chat(ctx)

// Горутина для отправки сообщений
go func() {
    for msg := range outbox {
        stream.Send(msg)
    }
    stream.CloseSend()
}()

// Горутина для получения сообщений
for {
    msg, err := stream.Recv()
    if err == io.EOF {
        break
    }
    displayMessage(msg)
}

// Сервер (Go)
func (s *chatServer) Chat(stream pb.ChatService_ChatServer) error {
    // Канал для новых сообщений от этого клиента
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        // Отправляем сообщение всем участникам чата
        for _, participant := range s.rooms[msg.RoomId] {
            participant.Send(msg)
        }
    }
}

// Unary: просто вызываем
resp, err := client.Unary(ctx, req)

// Server streaming: читаем в цикле
stream, err := client.ServerStream(ctx, req)
for {
    msg, err := stream.Recv()
    if err == io.EOF { break }
    // обрабатываем msg
}

// Client streaming: отправляем в цикле
stream, err := client.ClientStream(ctx)
for _, chunk := range chunks {
    stream.Send(chunk)
}
resp, err := stream.CloseAndRecv()

// Bidirectional: две горутины
stream, err := client.Bidirectional(ctx)
go sendMessages(stream)  // отправка
go receiveMessages(stream) // получение

// Unary: просто возвращаем
func (s *server) Unary(ctx context.Context, req *pb.Request) (*pb.Response, error) {
    return &pb.Response{}, nil
}

// Server streaming: отправляем в цикле
func (s *server) ServerStream(req *pb.Request, stream pb.Service_ServerStreamServer) error {
    for _, item := range items {
        stream.Send(item)
    }
    return nil
}

// Client streaming: читаем в цикле
func (s *server) ClientStream(stream pb.Service_ClientStreamServer) error {
    for {
        chunk, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.Response{})
        }
        // обрабатываем chunk
    }
}

// Bidirectional: читаем и отправляем одновременно
func (s *server) Bidirectional(stream pb.Service_BidirectionalServer) error {
    // Обработка входящих сообщений
    go func() {
        for {
            msg, err := stream.Recv()
            if err == io.EOF { return }
            // обрабатываем msg
        }
    }()
    
    // Отправка исходящих сообщений
    for msg := range outgoing {
        stream.Send(msg)
    }
    return nil
}

for {
    msg, err := stream.Recv()
    if err == io.EOF {
        break // поток закрыт корректно
    }
    if err != nil {
        // реальная ошибка
    }
}

stream.CloseSend()  // больше не будем отправлять
// Но всё ещё можем получать ответы

ctx, cancel := context.WithCancel(context.Background())
stream, err := client.Chat(ctx)

// Позже, когда нужно остановить
cancel()

service LogService {
    rpc TailLogs (TailRequest) returns (stream LogEntry);
}

message TailRequest {
    string service = 1;
    int32 lines = 2;
    string level = 3;
}

service PhotoService {
    rpc UploadPhoto (stream Chunk) returns (PhotoInfo);
}

message Chunk {
    bytes data = 1;
    int32 sequence = 2;
}

service GameService {
    rpc Play (stream PlayerAction) returns (stream GameState);
}

message PlayerAction {
    int32 player_id = 1;
    string action = 2;
    map<string, string> params = 3;
}

message GameState {
    repeated Player players = 1;
    repeated Bullet bullets = 2;
    int32 score = 3;
}

// Плохо: забыли CloseSend()
for _, chunk := range chunks {
    stream.Send(chunk)
}
resp, _ := stream.CloseAndRecv() // будет ждать вечно

// Плохо: игнорируем ошибки
for {
    msg, _ := stream.Recv()
    process(msg)
}

// Плохо: EOF как ошибка
for {
    msg, err := stream.Recv()
    if err != nil {
        break  // не отличаем EOF от реальной ошибки
    }
}

// Плохо: общий ресурс без синхронизации
var counter int
go func() {
    for { counter++ }  // пишем
}()
for { fmt.Println(counter) } // читаем