Primeiros passos com gRPC-Rust: streaming

1. Introdução

Neste codelab, você vai usar o gRPC-Rust para criar um cliente e um servidor que formam a base de um aplicativo de mapeamento de rotas escrito em Rust.

Ao final do tutorial, você terá um cliente que se conecta a um servidor remoto usando gRPC para receber informações sobre recursos em uma rota do cliente, criar um resumo da rota e trocar informações, como atualizações de trânsito, com o servidor e outros clientes.

O serviço é definido em um arquivo de buffers de protocolo, que será usado para gerar um código boilerplate para o cliente e o servidor, de modo que eles possam se comunicar entre si, economizando tempo e esforço na implementação dessa funcionalidade.

Esse código gerado cuida não apenas das complexidades da comunicação entre o servidor e o cliente, mas também da serialização e desserialização de dados.

O que você vai aprender

• Como usar buffers de protocolo para definir uma API de serviço.
• Como criar um cliente e um servidor baseados em gRPC a partir de uma definição de buffers de protocolo usando a geração de código automatizada.
• Uma compreensão da comunicação de streaming cliente-servidor com o gRPC.

Este codelab é destinado a desenvolvedores do Rust que são novos no gRPC ou que buscam uma atualização do gRPC, ou qualquer outra pessoa interessada em criar sistemas distribuídos. Não é necessário ter experiência anterior com o gRPC.

2. Antes de começar

Pré-requisitos

Instale o seguinte:

• GCC. Siga as instruções aqui.
• Git: instruções de instalação aqui.
• Rust, versão 1.88.0. Siga as instruções de instalação aqui.

Acessar o código

Para que você não precise começar do zero, este codelab fornece um scaffold do código-fonte do aplicativo para você concluir. As etapas a seguir mostram como finalizar o aplicativo, incluindo o uso dos plug-ins do compilador de buffers de protocolo para gerar o código boilerplate do gRPC.

Primeiro, crie o diretório de trabalho do codelab e cd nele:

mkdir streaming-grpc-rust-getting-started && cd streaming-grpc-rust-getting-started

Faça o download e extraia o codelab:

curl -sL https://github.com/grpc-ecosystem/grpc-codelabs/archive/refs/heads/2026.tar.gz \
  | tar xvz --strip-components=4 \
  grpc-codelabs-2026/codelabs/grpc-rust-streaming/start_here

Como alternativa, você pode fazer o download do arquivo .zip que contém apenas o diretório do codelab e descompactá-lo manualmente.

Fazer o download do ZIP

O código-fonte concluído está disponível no GitHub se você quiser pular a digitação de uma implementação.

3. Definir mensagens e serviços

A primeira etapa é definir o serviço gRPC do aplicativo, os métodos RPC e os tipos de mensagens de solicitação e resposta usando buffers de protocolo. Seu serviço vai fornecer:

• Métodos RPC chamados ListFeatures, RecordRoute e RouteChat que o servidor implementa e o cliente chama.
• Os tipos de mensagens Point, Feature, Rectangle, RouteNote e RouteSummary, que são estruturas de dados trocadas entre o cliente e o servidor ao chamar os métodos acima.

Esses métodos RPC e os tipos de mensagens serão definidos no arquivo proto/routeguide.proto do código-fonte fornecido.

Os buffers de protocolo são conhecidos como protobufs. Para mais informações sobre a terminologia do gRPC, consulte Conceitos principais, arquitetura e ciclo de vida do gRPC.

Definir tipos de mensagens

Primeiro, vamos definir as mensagens que serão usadas pelas RPCs. No arquivo proto/routeguide.proto do código-fonte, defina primeiro o tipo de mensagem Point. Um Point representa um par de coordenadas de latitude e longitude em um mapa. Para este codelab, use números inteiros para as coordenadas:

message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

Os números 1 e 2 são números de ID exclusivos para cada um dos campos na estrutura message.

Em seguida, defina o tipo de mensagem Feature. Um Feature usa um campo string para o nome ou endereço postal de algo em um local especificado por um Point:

message Feature {
  // The name or address of the feature.
  string name = 1;

  // The point where the feature is located.
  Point location = 2;
}

Em seguida, uma mensagem Rectangle que representa um retângulo de latitude-longitude, representado como dois pontos diagonalmente opostos "lo" e "hi".

message Rectangle {
  // One corner of the rectangle.
  Point lo = 1;

  // The other corner of the rectangle.
  Point hi = 2;
}

Também uma mensagem RouteNote que representa uma mensagem enviada em um determinado ponto.

message RouteNote {
  // The location from which the message is sent.
  Point location = 1;

  // The message to be sent.
  string message = 2;
}

Também exigimos uma mensagem RouteSummary. Essa mensagem é recebida em resposta a uma RPC RecordRoute, que é explicada na próxima seção. Ela contém o número de pontos individuais recebidos, o número de recursos detectados e a distância total percorrida como a soma cumulativa da distância entre cada ponto.

message RouteSummary {
  // The number of points received.
  int32 point_count = 1;

  // The number of known features passed while traversing the route.
  int32 feature_count = 2;

  // The distance covered in metres.
  int32 distance = 3;

  // The duration of the traversal in seconds.
  int32 elapsed_time = 4;
}

Definir métodos de serviço

Primeiro, vamos definir nosso serviço e, em seguida, definir nossas mensagens. Para definir um serviço, especifique um serviço nomeado no arquivo .proto. O arquivo proto/routeguide.proto tem uma estrutura service chamada RouteGuide que define um ou mais métodos fornecidos pelo serviço do aplicativo.

Defina métodos RPC na definição de serviço, especificando os tipos de solicitação e resposta. Nesta seção do codelab, vamos definir:

ListFeatures

Recebe os Features disponíveis no Rectangle especificado. Os resultados são transmitidos em vez de retornados de uma só vez (por exemplo, em uma mensagem de resposta com um campo repetido), já que o retângulo pode cobrir uma área grande e conter um grande número de recursos.

Um tipo adequado para essa RPC é uma RPC de streaming do lado do servidor: o cliente envia uma solicitação ao servidor e recebe um stream para ler uma sequência de mensagens. O cliente lê o stream retornado até que não haja mais mensagens. Como você pode ver no nosso exemplo, especifique um método de streaming do lado do servidor colocando a palavra-chave stream antes do tipo de resposta.

rpc ListFeatures(Rectangle) returns (stream Feature) {}

RecordRoute

Aceita um stream de Points em uma rota percorrida, retornando um RouteSummary quando o percurso é concluído.

Uma RPC de streaming do lado do cliente parece apropriada nesse caso: o cliente grava uma sequência de mensagens e as envia ao servidor, novamente usando um stream fornecido. Depois que o cliente termina de gravar as mensagens, ele aguarda que o servidor leia todas elas e retorne a resposta. Especifique um método de streaming do lado do cliente colocando a palavra-chave stream antes do tipo de solicitação.

rpc RecordRoute(stream Point) returns (RouteSummary) {}

RouteChat

Aceita um stream de RouteNotes enviados enquanto uma rota está sendo percorrida, enquanto recebe outros RouteNotes (por exemplo, de outros usuários).

Esse é exatamente o tipo de caso de uso para streaming bidirecional. Uma RPC de streaming bidirecional tem os dois lados enviando uma sequência de mensagens usando um stream de leitura e gravação. Os dois streams operam de maneira independente, para que clientes e servidores possam ler e gravar na ordem que quiserem.

Por exemplo, o servidor pode esperar para receber todas as mensagens do cliente antes de gravar as respostas ou pode ler uma mensagem e gravar uma mensagem ou alguma outra combinação de leituras e gravações.

A ordem das mensagens em cada stream é preservada. Especifique esse tipo de método colocando a palavra-chave stream antes da solicitação e da resposta.

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

4. Gerar o código do cliente e do servidor

Já fornecemos o código gerado do arquivo .proto no diretório generated/, incluindo todos os acréscimos feitos acima. No entanto, gostaríamos de explicar como a geração de código funciona.

Nosso arquivo .proto descreve todas as structs e funções que um cliente ou servidor usa. Usamos um script de build do Cargo (build.rs) com a caixa grpc-protobuf-build para gerar esse código automaticamente.

Em Cargo.toml, já adicionamos grpc-protobuf-build como uma dependência de build.

Em build.rs, configuramos grpc_protobuf_build::CodeGen para compilar proto/routeguide.proto no diretório generated/. As linhas principais estão aqui:

grpc_protobuf_build::CodeGen::new()
    .include("proto")
    .input("routeguide.proto")
    .output_dir("generated")
    .compile()
    .unwrap();

Isso chama a geração de código da caixa grpc_protobuf_build, transmitindo o routeguide.proto. Envolvemos isso em algum código para ser executado apenas quando um flag de recurso é transmitido, para que ele seja regenerado apenas quando você quiser. Não é necessário executar isso agora, já que geramos o código para você.

cargo build --bin routeguide-server --features regenerate_proto

Quando você executa o build do Cargo, build.rs compila as definições de buffer de protocolo no diretório generate/, incluindo:

• Definições de struct para tipos de mensagens Point e Feature.
• Um traço de serviço tônico que precisaremos implementar para o servidor: route_guide_server::RouteGuide.
• Um tipo de cliente gRPC-Rust que usaremos para chamar o servidor: route_guide_client::RouteGuideClient<T>.

Consulte o guia protoc-gen-rust-grpc para mais informações.

Em seguida, vamos implementar os métodos de serviço no servidor.

5. Implementar o serviço

Observação:uma implementação de servidor nativo usando o gRPC-Rust está em desenvolvimento. Enquanto isso, os servidores são implementados usando o Tonic com interoperabilidade total com clientes gRPC-Rust. Lembre-se: você pode implementar servidores e clientes de forma independente em qualquer linguagem, e eles são totalmente interoperáveis.

Primeiro, vamos conferir como criar um servidor RouteGuide. Há duas partes para fazer com que nosso serviço RouteGuide faça o trabalho:

• Implementar a interface de serviço gerada na nossa definição de serviço: fazer o "trabalho" real do nosso serviço.
• Executar um servidor gRPC para detectar solicitações de clientes e enviá-las para a implementação do método correto.

Em src/server/server.rs, podemos colocar o código gerado no escopo usando a macro include_generated_proto! do gRPC e importar o traço RouteGuide e Point.

mod grpc_pb {
    grpc::include_generated_proto!("generated", "routeguide");
}

pub use grpc_pb::{
    route_guide_server::{RouteGuideServer, RouteGuide},
    Point, Feature, Rectangle, RouteNote, RouteSummary
};

Podemos começar definindo uma struct para representar nosso serviço. Podemos fazer isso em src/server/server.rs por enquanto:

#[derive(Debug)]
pub struct RouteGuideService {
    features: Vec<Feature>,
}

Agora, precisamos implementar o traço route_guide_server::RouteGuide do nosso código gerado.

Implementar RouteGuide

Precisamos implementar a interface RouteGuide gerada. É assim que a implementação seria. Isso já está no modelo.

#[tonic::async_trait]
impl RouteGuide for RouteGuideService {
    async fn list_features(
        &self,
        request: Request<Rectangle>,
    ) -> Result<Response<ListFeaturesStream>, Status> {
        ...
    }

    async fn record_route(
        &self,
        request: Request<tonic::Streaming<Point>>,
    ) -> Result<Response<RouteSummary>, Status> {
        ...
    }

    async fn route_chat(
        &self,
        request: Request<tonic::Streaming<RouteNote>>,
    ) -> Result<Response<RouteChatStream>, Status> {
        ...
    }
}

Vamos analisar cada implementação de RPC em detalhes.

RPC de streaming do lado do servidor: ListFeatures

Vamos começar com ListFeatures. Essa é uma RPC de streaming do lado do servidor (o cliente vai enviar uma mensagem, o servidor vai responder com muitas), então precisamos enviar vários Features de volta ao nosso cliente.

async fn list_features(
        &self,
        request: Request<Rectangle>,
    ) -> Result<Response<ListFeaturesStream>, Status> {
    println!("ListFeatures = {:?}", request);

    let (tx, rx) = mpsc::channel(4);
    let features = self.features.clone();

    tokio::spawn(async move {
        for feature in &features[..] {
            if in_range(&feature.location().to_owned(), request.get_ref()) {
                println!("  => send {feature:?}");
                tx.send(Ok(feature.clone())).await.unwrap();
            }
        }
        println!(" /// done sending");
    });

    let output_stream = ReceiverStream::new(rx);
    Ok(Response::new(Box::pin(output_stream)))
}

Como você pode ver, recebemos um objeto de solicitação (o Rectangle em que nosso cliente quer encontrar Features). Desta vez, precisamos retornar um stream de valores. Criamos um canal e geramos uma nova tarefa assíncrona em que realizamos uma pesquisa, enviando os recursos que atendem às nossas restrições para o canal. A metade do stream do canal é retornada ao autor da chamada, encapsulada em uma tonic::Response.

RPC de streaming do lado do cliente: RecordRoute

Agora, vamos conferir algo um pouco mais complicado: o método de streaming do lado do cliente RecordRoute, em que recebemos um stream de Points do cliente e retornamos um único RouteSummary com informações sobre a viagem. Ele recebe um stream como entrada, que o servidor pode usar para ler e gravar mensagens. Ele pode iterar pelas mensagens do cliente usando o método next() e retornar a resposta única.

async fn record_route(
        &self,
        request: Request<tonic::Streaming<Point>>,
    ) -> Result<Response<RouteSummary>, Status> {
    println!("RecordRoute");
    let mut stream = request.into_inner();
    let mut summary = RouteSummary::default();
    let mut last_point = None;
    let now = Instant::now();

    while let Some(point) = stream.next().await {
        let point = point?;
        println!("  ==> Point = {point:?}");

        // Increment the point count
        summary.set_point_count(summary.point_count() + 1);

        // Find features
        for feature in &self.features[..] {
            if feature.location().latitude() == point.latitude() {
                if feature.location().longitude() == point.longitude(){
                    summary.set_feature_count(summary.feature_count() + 1);
                }
            }
        }

        // Calculate the distance
        if let Some(ref last_point) = last_point {
            let new_dist = summary.distance() + calc_distance(last_point, &point);
            summary.set_distance(new_dist);
        }
        last_point = Some(point);
    }
    summary.set_elapsed_time(now.elapsed().as_secs() as i32);
    Ok(Response::new(summary))
}

No corpo do método, usamos o método next() do stream para ler repetidamente as solicitações do cliente para um objeto de solicitação (neste caso, um Point) até que não haja mais mensagens. Se for "Nenhum", o stream ainda será bom e poderá continuar lendo.

RPC de streaming bidirecional: RouteChat

Por fim, vamos conferir nossa RPC de streaming bidirecional RouteChat().

async fn route_chat(
        &self,
        request: Request<tonic::Streaming<RouteNote>>,
    ) -> Result<Response<RouteChatStream>, Status> {
    println!("RouteChat");

    let mut notes: HashMap<(i32, i32), Vec<RouteNote>> = HashMap::new();
    let mut stream = request.into_inner();

    let output = async_stream::try_stream! {
        while let Some(note) = stream.next().await {
            let note = note?;
            let location = note.location();
            let key = (location.latitude(), location.longitude());
            let location_notes = notes.entry(key).or_insert(vec![]);
            location_notes.push(note);
            for note in location_notes {
                yield note.clone();
            }
        }
    };
    Ok(Response::new(Box::pin(output)))
}

Desta vez, recebemos um stream que, como no nosso exemplo de streaming do lado do cliente, pode ser usado para ler e gravar mensagens. No entanto, desta vez, retornamos valores pelo stream do nosso método enquanto o cliente ainda está gravando mensagens no stream de mensagens. A sintaxe para leitura e gravação aqui é muito semelhante ao nosso método de streaming do cliente, exceto que o servidor retorna um RouteChatStream. Embora cada lado sempre receba as mensagens do outro na ordem em que foram escritas, o cliente e o servidor podem ler e gravar em qualquer ordem. Os streams operam de maneira completamente independente.

Criamos o stream de saída usando try_stream!, que indica que o stream pode retornar erros.

Iniciar o servidor

Depois de implementar esse método, também precisamos iniciar um servidor gRPC para que os clientes possam usar nosso serviço. Preencha main().

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "[::1]:10000".parse().unwrap();
    println!("RouteGuideServer listening on: {addr}");
    let route_guide = RouteGuideService {
        features: load(),
    };
    let svc = RouteGuideServer::new(route_guide);
    Server::builder().add_service(svc).serve(addr).await?;
    Ok(())
}

Confira o que está acontecendo em main(), etapa por etapa:

1. Especifique a porta que queremos usar para detectar solicitações de clientes.
2. Crie um RouteGuideService com recursos carregados.
3. Crie uma instância do servidor gRPC usando RouteGuideServer::new() com o serviço que criamos.
4. Registre nossa implementação de serviço com o servidor gRPC.
5. Chame serve() no servidor com os detalhes da porta para fazer uma espera de bloqueio até que o processo seja encerrado.

6. Criar o cliente

Nesta seção, vamos conferir como criar um cliente Rust para nosso serviço RouteGuide em src/client/client.rs.

Primeiro, coloque o código gerado no escopo.

mod grpc_pb {
    grpc::include_generated_proto!("generated", "routeguide");
}

use grpc_pb::route_guide_client::RouteGuideClient;
use grpc_pb::{Point, Rectangle, RouteNote};

Chamar métodos de serviço

Agora, vamos conferir como chamar nossos métodos de serviço. No gRPC-Rust, as RPCs de streaming são assíncronas e não bloqueadoras, usando a sintaxe async/await do Rust e os streams do Tokio.

RPC de streaming do lado do servidor: PrintFeatures

Em RPCs de streaming do servidor, o cliente envia uma única mensagem de solicitação ao servidor e recebe um stream de mensagens de resposta. É aqui que, em client.rs, chamamos o método de streaming do lado do servidor list_features() (que corresponde à declaração de RPC ListFeatures encontrada no nosso proto). O servidor, por sua vez, vai enviar um stream de mensagens Feature:

async fn print_features(client: &RouteGuideClient<Channel>) -> Result<(), Box<dyn Error>> {
    let rectangle = proto!(Rectangle {
        lo: proto!(Point {
            latitude: 400_000_000,
            longitude: -750_000_000,
        }),
        hi: proto!(Point {
            latitude: 420_000_000,
            longitude: -730_000_000,
        }),
    });

    let mut stream = client.list_features(rectangle).await;

    while let Some(feature) = stream.recv().await {
        println!(
            "FEATURE: Name = \"{}\", Lat = {}, Lon = {}",
            feature.name(),
            feature.location().latitude(),
            feature.location().longitude()
        );
    }
    let status = stream.status().await;
    assert!(status.is_ok(), "{:?}", status);
    Ok(())
}

RPC de streaming do lado do cliente: RecordRoute

Quando usamos o streaming do lado do cliente, o cliente abre um stream para o servidor e envia uma sequência de mensagens. Ele vai receber uma única mensagem de resposta quando o stream terminar.

Aqui, iniciamos a chamada com client.record_route().await, enviamos várias coordenadas Point geradas uma por uma no stream usando stream.send(point).await e, em seguida, fechamos o stream com stream.close_and_recv().await para receber a única mensagem RouteSummary do servidor.

async fn run_record_route(client: &RouteGuideClient<Channel>) -> Result<(), Box<dyn Error>> {
    let mut rng = rand::rng();
    let point_count: i32 = rng.random_range(2..100);

    let mut points = vec![];
    for _ in 0..=point_count {
        points.push(random_point(&mut rng));
    }

    println!("Traversing {} points", points.len());
    let mut stream = client.record_route().await;

    for point in &points {
        if stream.send(point).await.is_err() {
            break;
        }
    }

    match stream.close_and_recv().await {
        Ok(response) => {
            println!(
                "SUMMARY: Feature Count = {}, Distance = {}",
                response.feature_count(),
                response.distance()
            );
        }
        Err(e) => println!("something went wrong: {e:?}"),
    }
    Ok(())
}

RPC de streaming bidirecional: RouteChat

Por fim, vamos conferir nossa RPC de streaming bidirecional RouteChat(). Aqui, o cliente e o servidor vão transmitir uma sequência de mensagens. Geramos uma tarefa tokio para enviar mensagens continuamente ao servidor com tx.send(note).await.is_err(). Enquanto isso, rx.recv().await detecta mensagens de resposta do servidor e as imprime à medida que chegam.

async fn run_route_chat(client: &RouteGuideClient<Channel>) -> Result<(), Box<dyn Error>> {
    let (mut tx, mut rx) = client.route_chat().await;

    let start = time::Instant::now();
    tokio::spawn(async move {
        let mut interval = time::interval(Duration::from_millis(50));
        for _ in 0..10 {
            let time = interval.tick().await;
            let elapsed = time.duration_since(start);
            let note = proto!(RouteNote {
                location: proto!(Point {
                    latitude: 409146138 + elapsed.as_millis() as i32,
                    longitude: -746188906,
                }),
                message: format!("at {elapsed:?}"),
            });
            if tx.send(note).await.is_err() {
                return;
            }
        }
        tx.close();
    });

    while let Some(note) = rx.recv().await {
        println!(
            "Note: Latitude = {}, Longitude = {}, Message = \"{}\"",
            note.location().latitude(),
            note.location().longitude(),
            note.message()
        );
    }
    let status = rx.status().await;
    assert!(status.is_ok(), "{:?}", status);
    Ok(())
}

Embora cada lado sempre receba as mensagens do outro na ordem em que foram escritas, o cliente e o servidor podem ler e gravar em qualquer ordem. Os streams operam de maneira completamente independente.

Criar e transmitir o cliente

Para chamar métodos de serviço, primeiro precisamos criar um canal para se comunicar com o servidor. Para isso, primeiro criamos um endpoint, nos conectamos a ele e transmitimos o canal criado quando conectado a RouteGuideClient::new() da seguinte maneira:

// Create channel to connect to server
let channel = Channel::builder(
    "dns:///[::1]:10000",
    Arc::new(LocalChannelCredentials::new()),
)
.build();

// Create a new client
let client = RouteGuideClient::new(channel);

Com esse cliente criado, podemos chamar os métodos que escrevemos acima, transmitindo o cliente para eles. Adicionamos todo esse código ao main(), que está usando o ambiente de execução assíncrono do Tokio. Confira o código completo:

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create channel to connect to server
    let channel = Channel::builder(
        "dns:///[::1]:10000",
        Arc::new(LocalChannelCredentials::new()),
    )
    .build();

    // Create a new client
    let client = RouteGuideClient::new(channel);

    println!("\n*** SERVER STREAMING ***");
    print_features(&client).await?;

    println!("\n*** CLIENT STREAMING ***");
    run_record_route(&client).await?;

    println!("\n*** BIDIRECTIONAL STREAMING ***");
    run_route_chat(&client).await?;

    Ok(())
}

7. Faça um teste

Para executar o cliente e o servidor, primeiro verifique se os dois destinos binários estão definidos em Cargo.toml:

[[bin]]
name = "routeguide-server"
path = "src/server/server.rs"

[[bin]]
name = "routeguide-client"
path = "src/client/client.rs"

Em seguida, execute os seguintes comandos no nosso diretório de trabalho:

1. Execute o servidor em um terminal:

cargo run --bin routeguide-server

2. Execute o cliente em outro terminal:

cargo run --bin routeguide-client

A resposta exibida será semelhante a esta:

*** SERVER STREAMING ***
FEATURE: Name = "Patriots Path, Mendham, NJ 07945, USA", Lat = 407838351, Lon = -746143763
FEATURE: Name = "101 New Jersey 10, Whippany, NJ 07981, USA", Lat = 408122808, Lon = -743999179
FEATURE: Name = "U.S. 6, Shohola, PA 18458, USA", Lat = 413628156, Lon = -749015468
...
*** CLIENT STREAMING ***
Traversing 86 points
SUMMARY: Feature Count = 0, Distance = 803709356

*** BIDIRECTIONAL STREAMING ***
Note: Latitude = 409146138, Longitude = -746188906, Message = "at 112.45µs"
Note: Latitude = 409146139, Longitude = -746188906, Message = "at 1.00011245s"
Note: Latitude = 409146140, Longitude = -746188906, Message = "at 2.00011245s"

8. A seguir

• Confira o repositório oficial do gRPC-Rust.
• Saiba mais sobre a arquitetura do gRPC em Conceitos principais.
• Consulte a documentação do gRPC-Rust em gRPC.io.

9. Colaboradores deste codelab

• Cathy Zhao
• Arvind Bright
• Nathaniel Ford