Getting Started with gRPC-Go - Streaming

1. Introduction

In this codelab, you'll use gRPC-Go to create a client and server that form the foundation of a route-mapping application written in Go.

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 Go 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:

  • The Go toolchain version 1.24.5 or later. For installation instructions, see Go's Getting started.
  • The protocol buffer compiler, protoc, version 3.27.1 or later. For installation instructions, see the compiler's installation guide.
  • The protocol buffer compiler plugins for Go and gRPC. To install these plugins, run the following commands:
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

Update your PATH variable so that the protocol buffer compiler can find the plugins:

export PATH="$PATH:$(go env GOPATH)/bin"

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-go-getting-started && cd streaming-grpc-go-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-go-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, and RouteChat that the server implements and the client calls.
  • The message types Point, Feature, Rectangle, RouteNote, and RouteSummary, 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 routeguide/route_guide.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

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

To define a service, you specify a named service in your .proto file. The route_guide.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 client and server code

Next, generate the boilerplate gRPC code for both the client and server from the .proto file using the protocol buffer compiler. In the routeguide directory, run:

protoc --go_out=. --go_opt=paths=source_relative \
       --go-grpc_out=. --go-grpc_opt=paths=source_relative \
       route_guide.proto

This command generates the following files:

  • route_guide.pb.go, which contains functions to create the application's message types and access their data and the definition of the types that represent the messages.
  • route_guide_grpc.pb.go, which contains functions the client uses to call the service's remote gRPC method, and functions used by the server to provide that remote service.

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 service implementation.

Let's implement RouteGuide in server/server.go.

Implement RouteGuide

We need to implement the generated RouteGuideService interface. This is how the implementation would look.

type routeGuideServer struct {
        ...
}
...
func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, stream pb.RouteGuide_ListFeaturesServer) error {
        ...
}
...

func (s *routeGuideServer) RecordRoute(stream pb.RouteGuide_RecordRouteServer) error {
        ...
}
...

func (s *routeGuideServer) RouteChat(stream pb.RouteGuide_RouteChatServer) error {
        ...
}

Let us look into each RPC implementation in detail.

Server-side streaming RPC

Start with one of our streaming RPCs. ListFeatures is a server-side streaming RPC, so we need to send back multiple Features to our client.

func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, stream pb.RouteGuide_ListFeaturesServer) error {
  for _, feature := range s.savedFeatures {
    if inRange(feature.Location, rect) {
      if err := stream.Send(feature); err != nil {
        return err
      }
    }
  }
  return nil
}

As you can see, instead of getting simple request and response objects in our method parameters, this time we get a request object (the Rectangle in which our client wants to find Features) and a special RouteGuide_ListFeaturesServer object to write our responses. In the method, we populate as many Feature objects as we need to return, writing them to the RouteGuide_ListFeaturesServer using its Send() method. Finally, as in our simple RPC, we return a nil error to tell gRPC that we've finished writing responses. Should any error happen in this call, we return a non-nil error; the gRPC layer will translate it into an appropriate RPC status to be sent on the wire.

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. As you can see, this time the method doesn't have a request parameter at all. Instead, it gets a RouteGuide_RecordRouteServer stream, which the server can use to both read and write messages - it can receive client messages using its Recv() method and return its single response using its SendAndClose() method.

func (s *routeGuideServer) RecordRoute(stream pb.RouteGuide_RecordRouteServer) error {
  var pointCount, featureCount, distance int32
  var lastPoint *pb.Point
  startTime := time.Now()
  for {
    point, err := stream.Recv()
    if err == io.EOF {
      endTime := time.Now()
      return stream.SendAndClose(&pb.RouteSummary{
        PointCount:   pointCount,
        FeatureCount: featureCount,
        Distance:     distance,
        ElapsedTime:  int32(endTime.Sub(startTime).Seconds()),
      })
    }
    if err != nil {
      return err
    }
    pointCount++
    for _, feature := range s.savedFeatures {
      if proto.Equal(feature.Location, point) {
        featureCount++
      }
    }
    if lastPoint != nil {
      distance += calcDistance(lastPoint, point)
    }
    lastPoint = point
  }
}

In the method body we use the RouteGuide_RecordRouteServer's Recv() method to repeatedly read in our client's requests to a request object (in this case a Point) until there are no more messages: the server needs to check the error returned from Recv() after each call. If this is nil, the stream is still good and it can continue reading; if it's io.EOF the message stream has ended and the server can return its RouteSummary. If it has any other value, we return the error "as is" so that it'll be translated to an RPC status by the gRPC layer.

Bidirectional streaming RPC

Finally, let's look at our bidirectional streaming RPC RouteChat().

func (s *routeGuideServer) RouteChat(stream pb.RouteGuide_RouteChatServer) error {
  for {
    in, err := stream.Recv()
    if err == io.EOF {
      return nil
    }
    if err != nil {
      return err
    }
    key := serialize(in.Location)

    s.mu.Lock()
    s.routeNotes[key] = append(s.routeNotes[key], in)
    // Note: this copy prevents blocking other clients while serving this one.
    // We don't need to do a deep copy, because elements in the slice are
    // insert-only and never modified.
    rn := make([]*pb.RouteNote, len(s.routeNotes[key]))
    copy(rn, s.routeNotes[key])
    s.mu.Unlock()

    for _, note := range rn {
      if err := stream.Send(note); err != nil {
        return err
      }
    }
  }
}

This time we get a RouteGuide_RouteChatServer 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 uses the stream's send() method rather than SendAndClose() because it's writing multiple responses. 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.

Start the server

Once we've implemented all our methods, we also need to start up a gRPC server so that clients can actually use our service. The following snippet shows how we do this for our RouteGuide service:

lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", *port))
if err != nil {
  log.Fatalf("failed to listen: %v", err)
}

grpcServer := grpc.NewServer()

s := &routeGuideServer{routeNotes: make(map[string][]*pb.RouteNote)}
s.loadFeatures()
pb.RegisterRouteGuideServer(grpcServer, s)
grpcServer.Serve(lis)

Here's what is happening in main(), step by step:

  1. Specify the TCP port to use to listen for remote client requests, using lis, err := net.Listen(...). By default, the application uses TCP port 50051 as specified by the variable port or by passing the --port switch on the command line when running the server. If the TCP port can't be opened, the application ends with a fatal error.
  2. Create an instance of the gRPC server using grpc.NewServer(...), naming this instance grpcServer.
  3. Create a pointer to routeGuideServer, a structure representing the application's API service, naming the pointer s.
  4. Use s.loadFeatures() to populate the array s.savedFeatures.
  5. Register our service implementation with the gRPC server.
  6. Call Serve() on the server with our port details to do a blocking wait for client requests; this continues until the process is killed or Stop() is called.

The function loadFeatures() gets its coordinates-to-location mappings from server/testdata.go.

6. Create the client

Now edit client/client.go, which is where you'll implement the client code.

To call the remote service's methods, we first need to create a gRPC channel to communicate with the server. We create this by passing the server's target URI string (which in this case is simply the address and port number) to grpc.NewClient() in the client's main() function as follows:

// Set up a connection to the gRPC server.
conn, err := grpc.NewClient("dns:///"+*serverAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
  log.Fatalf("fail to dial: %v", err)
}
defer conn.Close()

The server's address, defined by the variable serverAddr, is by default localhost:50051, and can be overridden by the --addr switch on the command line when running the client.

If the client needs to connect to a service that requires authentication credentials, such as TLS or JWT credentials, the client can pass a DialOptions object as a parameter to grpc.NewClient that contains the required credentials. The RouteGuide service doesn't require any credentials.

Once the gRPC channel is set up, we need a client stub to perform RPCs via Go function calls. We get that stub using the NewRouteGuideClient method provided by the route_guide_grpc.pb.go file generated from the application's .proto file.

import (pb "github.com/grpc-ecosystem/codelabs/getting_started_streaming/routeguide")

client := pb.NewRouteGuideClient(conn)

Call service methods

Now let's look at how we call our service methods. In gRPC-Go, 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.

rect := &pb.Rectangle{ ... }  // initialize a pb.Rectangle
log.Printf("Looking for features within %v", rect)
stream, err := client.ListFeatures(context.Background(), rect)
if err != nil {
  log.Fatalf("client.ListFeatures failed: %v", err)
}
for {
  // For server-to-client streaming RPCs, you call stream.Recv() until it
  // returns io.EOF.
  feature, err := stream.Recv()
  if err == io.EOF {
    break
  }
  if err != nil {
    log.Fatalf("client.ListFeatures failed: %v", err)
  }
  log.Printf("Feature: name: %q, point:(%v, %v)", feature.GetName(),
    feature.GetLocation().GetLatitude(), feature.GetLocation().GetLongitude())
}

As in the simple RPC, we pass the method a context and a request. However, instead of getting a response object back, we get back an instance of RouteGuide_ListFeaturesClient. The client can use the RouteGuide_ListFeaturesClient stream to read the server's responses. We use the RouteGuide_ListFeaturesClient's Recv() 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: the client needs to check the error err returned from Recv() after each call. If nil, the stream is still good and it can continue reading; if it's io.EOF then the message stream has ended; otherwise there must be an RPC error, which is passed over through err.

Client-side streaming RPC

The client-side streaming method RecordRoute is similar to the server-side method, except that we only pass the method a context and get a RouteGuide_RecordRouteClient stream back, which we can use to both write and read messages.

// Create a random number of random points
r := rand.New(rand.NewSource(time.Now().UnixNano()))
pointCount := int(r.Int31n(100)) + 2 // Traverse at least two points
var points []*pb.Point
for i := 0; i < pointCount; i++ {
  points = append(points, randomPoint(r))
}
log.Printf("Traversing %d points.", len(points))
c2sStream, err := client.RecordRoute(context.TODO())
if err != nil {
  log.Fatalf("client.RecordRoute failed: %v", err)
}
// Stream each point to the server.
for _, point := range points {
  if err := c2sStream.Send(point); err != nil {
    log.Fatalf("client.RecordRoute: stream.Send(%v) failed: %v", point, err)
  }
}
// Close the stream and receive the RouteSummary from the server.
reply, err := c2sStream.CloseAndRecv()
if err != nil {
  log.Fatalf("client.RecordRoute failed: %v", err)
}
log.Printf("Route summary: %v", reply)

The RouteGuide_RecordRouteClient has a Send() method that we can use to send requests to the server. Once we've finished writing our client's requests to the stream using Send(), we need to call CloseAndRecv() on the stream to let gRPC know that we've finished writing and are expecting to receive a response. We get our RPC status from the err returned from CloseAndRecv(). If the status is nil, then the first return value from CloseAndRecv() will be a valid server response.

Bidirectional streaming RPC

Finally, let's look at our bidirectional streaming RPC RouteChat(). As in the case of RecordRoute, we only pass the method a context object and get back a stream that we can use to both write and read messages. However, this time we return values via our method's stream while the server is still writing messages to their message stream.

biDiStream, err := client.RouteChat(context.Background())
if err != nil {
  log.Fatalf("client.RouteChat failed: %v", err)
}
// this channel is used to wait for the receive goroutine to finish.
recvDoneCh := make(chan struct{})
// receive goroutine.
go func() {
  for {
    in, err := biDiStream.Recv()
    if err == io.EOF {
      // read done.
      close(recvDoneCh)
      return
    }
    if err != nil {
      log.Fatalf("client.RouteChat failed: %v", err)
    }
    log.Printf("Got message %s at point(%d, %d)", in.Message, in.Location.Latitude, in.Location.Longitude)
  }
}()
// send messages simultaneously.
for _, note := range notes {
  if err := biDiStream.Send(note); err != nil {
    log.Fatalf("client.RouteChat: stream.Send(%v) failed: %v", note, err)
  }
}
biDiStream.CloseSend()
// wait for the receive goroutine to finish.
<-recvDoneCh

The syntax for reading and writing here is very similar to our client-side streaming method, except we use the stream's CloseSend() method once we've finished our call. 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.

7. Try it out

Confirm the server and client are working with each other correctly by executing the following commands in the application's working directory:

  1. Run the server in one terminal:
cd server
go run .
  1. Run the client from another terminal:
cd client
go run .

You'll see output like this, with timestamps omitted for clarity:

Looking for features within lo:<latitude:400000000 longitude:-750000000 > hi:<latitude:420000000 longitude:-730000000 >
name:"Patriots Path, Mendham, NJ 07945, USA" location:<latitude:407838351 longitude:-746143763 >
...
name:"3 Hasta Way, Newton, NJ 07860, USA" location:<latitude:410248224 longitude:-747127767 >
Traversing 56 points.
Route summary: point_count:56 distance:497013163
Got message First message at point(0, 1)
Got message Second message at point(0, 2)
Got message Third message at point(0, 3)
Got message First message at point(0, 1)
Got message Fourth message at point(0, 1)
Got message Second message at point(0, 2)
Got message Fifth message at point(0, 2)
Got message Third message at point(0, 3)
Got message Sixth message at point(0, 3)

8. What's next