1. Introduction
In this codelab, you'll use gRPC-Rust to create a client and server that form the foundation of a route-mapping application written in Rust.
By the end of the tutorial, you will have a client that connects to a remote server using gRPC to get information about features on a client's route, create a summary of a client's route, and exchange route information such as traffic updates with the server and other clients.
The service is defined in a Protocol Buffers file, which will be used to generate boilerplate code for the client and server so that they can communicate with each other, saving you time and effort in implementing that functionality.
This generated code takes care of not only the complexities of the communication between the server and client, but also data serialization and deserialization.
What you'll learn
- How to use Protocol Buffers to define a service API.
- How to build a gRPC-based client and server from a Protocol Buffers definition using automated code generation.
- An understanding of client-server streaming communication with gRPC.
This codelab is aimed at Rust developers new to gRPC or seeking a refresher of gRPC, or anyone else interested in building distributed systems. No prior gRPC experience is required.
2. Before you begin
Prerequisites
Make sure you have installed the following:
Get the code
So that you don't have to start entirely from scratch, this codelab provides a scaffold of the application's source code for you to complete. The following steps will show you how to finish the application, including using the protocol buffer compiler plugins to generate the boilerplate gRPC code.
First, create the codelab working directory and cd into it:
mkdir streaming-grpc-rust-getting-started && cd streaming-grpc-rust-getting-started
Download and extract the codelab:
curl -sL https://github.com/grpc-ecosystem/grpc-codelabs/archive/refs/heads/v1.tar.gz \ | tar xvz --strip-components=4 \ grpc-codelabs-1/codelabs/grpc-rust-streaming/start_here
Alternatively, you can download the .zip file containing only the codelab directory and manually unzip it.
The completed source code is available on GitHub if you want to skip typing in an implementation.
3. Define messages and services
Your first step is to define the application's gRPC service, its RPC methods, and its request and response message types using Protocol Buffers. Your service will provide:
- RPC methods called
ListFeatures,RecordRoute, andRouteChatthat the server implements and the client calls. - The message types
Point,Feature,Rectangle,RouteNote, andRouteSummary, which are data structures exchanged between the client and server when calling methods above.
These RPC methods and its message types will all be defined in the proto/routeguide.proto file of the provided source code.
Protocol Buffers are commonly known as protobufs. For more information on gRPC terminology, see gRPC's Core concepts, architecture, and lifecycle.
Define Message types
Let's first define our messages that will be used by our RPCs. In the routeguide/route_guide.proto file of the source code, first define the Point message type. A Point represents a latitude-longitude coordinate pair on a map. For this codelab, use integers for the coordinates:
message Point {
int32 latitude = 1;
int32 longitude = 2;
}
The numbers 1 and 2 are unique ID numbers for each of the fields in the message structure.
Next, define the Feature message type. A Feature uses a string field for the name or postal address of something at a location specified by a Point:
message Feature {
// The name or address of the feature.
string name = 1;
// The point where the feature is located.
Point location = 2;
}
Next a Rectangle message which represents a latitude-longitude rectangle, represented as two diagonally opposite points "lo" and "hi".
message Rectangle {
// One corner of the rectangle.
Point lo = 1;
// The other corner of the rectangle.
Point hi = 2;
}
Also a RouteNote message which represents a message sent while at a given point.
message RouteNote {
// The location from which the message is sent.
Point location = 1;
// The message to be sent.
string message = 2;
}
We would also require a RouteSummary message. This message is received in response to a RecordRoute RPC which is explained in the next section. It contains the number of individual points received, the number of detected features, and the total distance covered as the cumulative sum of the distance between each point.
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;
}
Define service methods
Let's first define our service and then define our messages later. To define a service, you specify a named service in your .proto file. The proto/routeguide.proto file has a service structure named RouteGuide that defines one or more methods provided by the application's service.
Define RPC methods inside your service definition, specifying their request and response types. In this section of the codelab, let's define:
ListFeatures
Obtains the Features available within the given Rectangle. Results are streamed rather than returned at once (e.g. in a response message with a repeated field), as the rectangle may cover a large area and contain a huge number of features.
An appropriate type for this RPC is a server-side streaming RPC: the client sends a request to the server and gets a stream to read a sequence of messages back. The client reads from the returned stream until there are no more messages. As you can see in our example, you specify a server-side streaming method by placing the stream keyword before the response type.
rpc ListFeatures(Rectangle) returns (stream Feature) {}
RecordRoute
Accepts a stream of Points on a route being traversed, returning a RouteSummary when traversal is completed.
A client-side streaming RPC seems appropriate in this case: the client writes a sequence of messages and sends them to the server, again using a provided stream. Once the client has finished writing the messages, it waits for the server to read them all and return its response. You specify a client-side streaming method by placing the stream keyword before the request type.
rpc RecordRoute(stream Point) returns (RouteSummary) {}
RouteChat
Accepts a stream of RouteNotes sent while a route is being traversed, while receiving other RouteNotes (e.g. from other users).
This is exactly the kind of use case for bidirectional streaming. A bidirectional streaming RPC has both sides send a sequence of messages using a read-write stream. The two streams operate independently, so clients and servers can read and write in whatever order they like.
For example, the server could wait to receive all the client messages before writing its responses, or it could alternately read a message then write a message, or some other combination of reads and writes.
The order of messages in each stream is preserved. You specify this type of method by placing the stream keyword before both the request and the response.
rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}
4. Generate the client and server code
We have already given you the generated code from the .proto file in the generated directory.
If you want to learn how to generate code from the .proto file yourself or make any changes to the .proto file and test them out, refer to these instructions.
The generated code contains:
- Struct definitions for message types
Point,Feature,Rectangle,RouteNote, andRouteSummary. - A service trait we'll need to implement:
route_guide_server::RouteGuide. - A client type we'll use to call the server:
route_guide_client::RouteGuideClient<T>.
Next, we'll implement the methods on the server-side, so that when the client sends a request, the server can reply with an answer.
5. Implement the service
First let's look at how we create a RouteGuide server. There are two parts to making our RouteGuide service do its job:
- Implementing the service interface generated from our service definition: doing the actual "work" of our service.
- Running a gRPC server to listen for requests from clients and dispatch them to the right method implementation.
In src/server/server.rs, we can bring the generated code into scope through gRPC's include_generated_proto! macro and import the RouteGuide trait and Point.
mod grpc_pb {
grpc::include_generated_proto!("generated", "routeguide");
}
pub use grpc_pb::{
route_guide_server::{RouteGuideServer, RouteGuide},
Point, Feature, Rectangle, RouteNote, RouteSummary
};
We can start by defining a struct to represent our service. We can do this on src/server/server.rs for now:
#[derive(Debug)]
pub struct RouteGuideService {
features: Vec<Feature>,
}
Now, we need to implement the route_guide_server::RouteGuide trait from our generated code.
Implement RouteGuide
We need to implement the generated RouteGuide interface. This is how the implementation would look. This is already in the template.
#[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> {
...
}
}
Let us look into each RPC implementation in detail.
Server-side streaming RPC
Let's start with ListFeatures. This is a server-side streaming RPC, so we need to send back multiple Features to our client.
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)))
}
As you can see, we get a request object (the Rectangle in which our client wants to find Features). This time, we need to return a stream of values. We create a channel and spawn a new asynchronous task where we perform a lookup, sending the features that satisfy our constraints into the channel. The Stream half of the channel is returned to the caller, wrapped in a tonic::Response.
Client-side streaming RPC
Now let's look at something a little more complicated: the client-side streaming method RecordRoute, where we get a stream of Points from the client and return a single RouteSummary with information about their trip. It gets a stream as an input, which the server can use to both read and write messages. It can iterate through client messages using its next() method and return its single response.
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))
}
In the method body, we use the stream's next() method to repeatedly read in our client's requests to a request object (in this case a Point) until there are no more messages. If this is None, the stream is still good and it can continue reading.
Bidirectional streaming RPC
Finally, let's look at our bidirectional streaming RPC 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)))
}
This time we get a stream that, as in our client-side streaming example, can be used to read and write messages. However, this time we return values via our method's stream while the client is still writing messages to their message stream. The syntax for reading and writing here is very similar to our client-streaming method, except the server returns a RouteChatStream. Although each side will always get the other's messages in the order they were written, both the client and server can read and write in any order — the streams operate completely independently.
We create the output stream using try_stream!, which indicates that the stream could return errors.
Start the server
Once we've implemented this method, we also need to start up a gRPC server so that clients can actually use our service. Fill in 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(())
}
Here's what is happening in main(), step by step:
- Specify the port we want to use to listen for client requests
- Create a
RouteGuideServicewith features loaded in - Create an instance of the gRPC server using
RouteGuideServer::new()using the service we created. - Register our service implementation with the gRPC server.
- Call
serve()on the server with our port details to do a blocking wait until the process is killed.
6. Create the client
In this section, we'll look at creating a Rust client for our RouteGuide service in src/client/client.rs.
First, bring the generated code into scope.
mod grpc_pb {
grpc::include_generated_proto!("generated", "routeguide");
}
use grpc_pb::route_guide_client::RouteGuideClient;
use grpc_pb::{Point, Rectangle, RouteNote};
Call service methods
Now let's look at how we call our service methods. In gRPC-Rust, RPCs operate in a blocking/synchronous mode, which means that the RPC call waits for the server to respond, and will either return a response or an error.
Server-side streaming RPC
Here's where we call the server-side streaming method ListFeatures, which returns a stream of geographical Feature objects.
async fn print_features(client: &mut 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(Request::new(rectangle))
.await?
.into_inner();
while let Some(feature) = stream.message().await? {
println!("FEATURE: Name = \"{}\", Lat = {}, Lon = {}",
feature.name(),
feature.location().latitude(),
feature.location().longitude());
}
Ok(())
}
We pass the method a request and get back an instance of ListFeaturesStream. The client can use the ListFeaturesStream stream to read the server's responses. We use the ListFeaturesStream's message() method to repeatedly read in the server's responses to a response protocol buffer object (in this case a Feature) until there are no more messages.
Client-side streaming RPC
Here for record_route, we turn a vector of points into a stream. We then pass this stream into record_route() as a request and get a single RouteSummary response after the stream has been fully processed by the server.
async fn run_record_route(client: &mut 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 request = Request::new(tokio_stream::iter(points));
match client.record_route(request).await {
Ok(response) => {
let response = response.into_inner();
println!("SUMMARY: Feature Count = {}, Distance = {}", response.feature_count(), response.distance())},
Err(e) => println!("something went wrong: {e:?}"),
}
Ok(())
}
Bidirectional streaming RPC
Finally, let's look at our bidirectional streaming RPC RouteChat(). We pass the method a stream request that we write to and get back a stream that we can read messages from. This time we return values via our method's stream while the server is still writing messages to their message stream.
async fn run_route_chat(client: &mut RouteGuideClient<Channel>) -> Result<(), Box<dyn Error>> {
let start = time::Instant::now();
let outbound = async_stream::stream! {
let mut interval = time::interval(Duration::from_secs(1));
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_secs() as i32,
longitude: -746188906,
}),
message: format!("at {elapsed:?}"),
});
yield note;
}
};
let response = client.route_chat(Request::new(outbound)).await?;
let mut inbound = response.into_inner();
while let Some(note) = inbound.message().await? {
println!("Note: Latitude = {}, Longitude = {}, Message = \"{}\"",
note.location().latitude(),
note.location().longitude(),
note.message());
}
Ok(())
}
Although each side will always get the other's messages in the order they were written, both the client and server can read and write in any order — the streams operate completely independently.
Call helper methods
To call service methods, we first need to create a channel to communicate with the server. We create this by first creating an endpoint, connecting to that endpoint, and passing the channel made when connected to RouteGuideClient::new() as follows:
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Create endpoint to connect to
let endpoint = Endpoint::new("http://[::1]:10000")?;
let channel = endpoint.connect().await?;
// Create a new client
let mut client = RouteGuideClient::new(channel);
Ok(())
}
In main(), execute the methods we just created.
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Create endpoint to connect to
let endpoint = Endpoint::new("http://[::1]:10000")?;
let channel = endpoint.connect().await?;
// Create a new client
let mut client = RouteGuideClient::new(channel);
println!("\n*** SERVER STREAMING ***");
print_features(&mut client).await?;
println!("\n*** CLIENT STREAMING ***");
run_record_route(&mut client).await?;
println!("\n*** BIDIRECTIONAL STREAMING ***");
run_route_chat(&mut client).await?;
Ok(())
}
7. Try it out
First, to run our Client and Server, let's add them as binary targets to our crate. We need to edit our Cargo.toml accordingly:
[[bin]]
name = "routeguide-server"
path = "src/server/server.rs"
[[bin]]
name = "routeguide-client"
path = "src/client/client.rs"
As with any project, we also need to think of the dependencies that are necessary for our code to work. For Rust projects, the dependencies will be in Cargo.toml. We have already listed the necessary dependencies in the Cargo.toml file.
Then, execute the following commands from our working directories:
- Run the server in one terminal:
RUSTFLAGS="-Awarnings" cargo run --bin routeguide-server
- Run the client from another terminal:
RUSTFLAGS="-Awarnings" cargo run --bin routeguide-client
You'll see output like this:
*** 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. What's next
- Learn how gRPC works in Introduction to gRPC and Core concepts.
- Work through the Basics tutorial.