Getting Started with gRPC-Java - Streaming

1. Introduction

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

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 Java 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

  • JDK version 24.

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-java-getting-started && cd streaming-grpc-java-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-java-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 method, 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.

Protocol Buffers are commonly known as protobufs. For more information on gRPC terminology, see gRPC's Core concepts, architecture, and lifecycle.

This RPC method and its message types will all be defined in the proto/routeguide/route_guide.proto file of the provided source code.

Let's create a route_guide.proto file.

As we're generating Java code in this example, we've specified a java_package file option in our .proto:

option java_package = "io.grpc.examples.routeguide";
option java_outer_classname = "RouteGuideProto";

Define message types

In the proto/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;
}

So that multiple points within an area can be streamed to a client, you'll need a Rectangle message that 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 that 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;
}

Finally, you'll need 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.

When you define RPC methods inside your service definition, you specify their request and response types. In this section of the codelab, let's define:

ListFeatures

Obtains the Feature objects available within the given Rectangle. Results are streamed rather than returned at once as the rectangle may cover a large area and contain a huge number of features.

For this application, you'll use 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 is 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 usecase for bidirectional streaming. A bidirectional streaming RPC where 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 we need to generate the gRPC client and server interfaces from our .proto service definition. We do this using the protocol buffer compiler protoc with a special gRPC Java plugin. You need to use the proto3 compiler (which supports both proto2 and proto3 syntax) in order to generate gRPC services.

When using Gradle or Maven, the protoc build plugin can generate the necessary code as part of the build. You can refer to the grpc-java README for how to generate code from your own .proto files.

We have provided Gradle configuration.

From the streaming-grpc-java-getting-started directory enter

$ chmod +x gradlew
$ ./gradlew generateProto

The following classes are generated from our service definition (under build/generated/sources/proto/main/java):

  • One for each message type: Feature.java, Rectangle.java, ... which contain all of the protocol buffer code to populate, serialize, and retrieve our request and response message types.
  • RouteGuideGrpc.java which contains (along with some other useful code) a base class for RouteGuide servers to implement, RouteGuideGrpc.RouteGuideImplBase, with all the methods defined in the RouteGuide service and stub classes for clients to use

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.

Implement RouteGuide

We will implement a RouteGuideService class will extend the generated RouteGuideGrpc.RouteGuideImplBase class. This is how the implementation would look.

public void listFeatures(Rectangle request, StreamObserver<Feature> responseObserver) {
        ...
}

public StreamObserver<Point> recordRoute(final StreamObserver<RouteSummary> responseObserver) {

        ...
}

public StreamObserver<RouteNote> routeChat(final StreamObserver<RouteNote> responseObserver) {

        ...
}

Let us look into each RPC implementation in detail

Server-side streaming RPC

Next let's look at one of our streaming RPCs. ListFeatures is a server-side streaming RPC, so we need to send back multiple Features to our client.

private final Collection<Feature> features;

@Override
public void listFeatures(Rectangle request, StreamObserver<Feature> responseObserver) {
  int left = min(request.getLo().getLongitude(), request.getHi().getLongitude());
  int right = max(request.getLo().getLongitude(), request.getHi().getLongitude());
  int top = max(request.getLo().getLatitude(), request.getHi().getLatitude());
  int bottom = min(request.getLo().getLatitude(), request.getHi().getLatitude());

  for (Feature feature : features) {
    if (!RouteGuideUtil.exists(feature)) {
      continue;
    }

    int lat = feature.getLocation().getLatitude();
    int lon = feature.getLocation().getLongitude();
    if (lon >= left && lon <= right && lat >= bottom && lat <= top) {
      responseObserver.onNext(feature);
    }
  }
  responseObserver.onCompleted();
}

Like the simple RPC, this method gets a request object (the Rectangle in which our client wants to find Features) and a StreamObserver response observer.

This time, we get as many Feature objects as we need to return to the client (in this case, we select them from the service's feature collection based on whether they're inside our request Rectangle), and write them each in turn to the response observer using its onNext() method. Finally, as in our simple RPC, we use the response observer's onCompleted() method to tell gRPC that we've finished writing responses.

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.

@Override
public StreamObserver<Point> recordRoute(final StreamObserver<RouteSummary> responseObserver) {
  return new StreamObserver<Point>() {
    int pointCount;
    int featureCount;
    int distance;
    Point previous;
    long startTime = System.nanoTime();

    @Override
    public void onNext(Point point) {
      pointCount++;
      if (RouteGuideUtil.exists(checkFeature(point))) {
        featureCount++;
      }
      // For each point after the first, add the incremental distance from the previous point
      // to the total distance value.
      if (previous != null) {
        distance += calcDistance(previous, point);
      }
      previous = point;
    }

    @Override
    public void onError(Throwable t) {
      logger.log(Level.WARNING, "Encountered error in recordRoute", t);
    }

    @Override
    public void onCompleted() {
      long seconds = NANOSECONDS.toSeconds(System.nanoTime() - startTime);
      responseObserver.onNext(RouteSummary.newBuilder().setPointCount(pointCount)
          .setFeatureCount(featureCount).setDistance(distance)
          .setElapsedTime((int) seconds).build());
      responseObserver.onCompleted();
    }
  };
}

As you can see, like the previous method types our method gets a StreamObserver responseObserver parameter, but this time it returns a StreamObserver for the client to write its Points.

In the method body we instantiate an anonymous StreamObserver to return, in which we:

  • Override the onNext() method to get features and other information each time the client writes a Point to the message stream.
  • Override the onCompleted() method (called when the client has finished writing messages) to populate and build our RouteSummary. We then call our method's own response observer's onNext() with our RouteSummary, and then call its onCompleted() method to finish the call from the server side.

Bidirectional streaming RPC

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

@Override
public StreamObserver<RouteNote> routeChat(final StreamObserver<RouteNote> responseObserver) {
  return new StreamObserver<RouteNote>() {
    @Override
    public void onNext(RouteNote note) {
      List<RouteNote> notes = getOrCreateNotes(note.getLocation());

      // Respond with all previous notes at this location.
      for (RouteNote prevNote : notes.toArray(new RouteNote[0])) {
        responseObserver.onNext(prevNote);
      }

      // Now add the new note to the list
      notes.add(note);
    }

    @Override
    public void onError(Throwable t) {
      logger.log(Level.WARNING, "Encountered error in routeChat", t);
    }

    @Override
    public void onCompleted() {
      responseObserver.onCompleted();
    }
  };
}

As with our client-side streaming example, we both get and return a StreamObserver, except this time we return values via our method's response observer while the client is still writing messages to their message stream. The syntax for reading and writing here is exactly the same as for our client-streaming and server-streaming methods. 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:

public RouteGuideServer(int port, URL featureFile) throws IOException {
  this(ServerBuilder.forPort(port), port, RouteGuideUtil.parseFeatures(featureFile));
}

/** Create a RouteGuide server using serverBuilder as a base and features as data. */
public RouteGuideServer(ServerBuilder<?> serverBuilder, int port, Collection<Feature> features) {
  this.port = port;
  server = serverBuilder.addService(new RouteGuideService(features))
      .build();
}

public void start() throws IOException {
  server.start();
  logger.info("Server started, listening on " + port);
}

As you can see, we build and start our server using a ServerBuilder.

To do this, we:

  1. Specify the address and port we want to use to listen for client requests using the builder's forPort() method.
  2. Create an instance of our service implementation class RouteGuideService and pass it to the builder's addService() method.
  3. Call build() and start() on the builder to create and start an RPC server for our service.

Since the ServerBuilder already incorporates the port, the only reason we pass a port is to use it for logging.

6. Create the client

In this section, we'll look at creating a client for our RouteGuide service. You can see our complete example client code in ../complete/src/main/java/io/grpc/complete/routeguide/ RouteGuideClient.java.

Instantiate a stub

To call service methods, we first need to create a stub, or rather, two stubs:

  • a blocking/synchronous stub: this means that the RPC call waits for the server to respond, and will either return a response or raise an exception.
  • a non-blocking/asynchronous stub that makes non-blocking calls to the server, where the response is returned asynchronously. You can make certain types of streaming calls only by using an asynchronous stub.

First we need to create a gRPC channel for our stub, specifying the server address and port we want to connect to:

  public static void main(String[] args) throws InterruptedException {
    String target = "localhost:8980";
    if (args.length > 0) {
      if ("--help".equals(args[0])) {
        System.err.println("Usage: [target]");
        System.err.println("");
        System.err.println("  target  The server to connect to. Defaults to " + target);
        System.exit(1);
      }
      target = args[0];
    }

    List<Feature> features;
    try {
      features = RouteGuideUtil.parseFeatures(RouteGuideUtil.getDefaultFeaturesFile());
    } catch (IOException ex) {
      ex.printStackTrace();
      return;
    }

    ManagedChannel channel = Grpc.newChannelBuilder(target, InsecureChannelCredentials.create())
        .build();
    try {
      RouteGuideClient client = new RouteGuideClient(channel);

      // Looking for features between 40, -75 and 42, -73.
      client.listFeatures(400000000, -750000000, 420000000, -730000000);

      // Record a few randomly selected points from the features file.
      client.recordRoute(features, 10);

      // Send and receive some notes.
      CountDownLatch finishLatch = client.routeChat();

      if (!finishLatch.await(1, TimeUnit.MINUTES)) {
        client.warning("routeChat did not finish within 1 minutes");
      }
    } finally {
      channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
    }
  }

We use a ManagedChannelBuilder to create the channel.

Now we can use the channel to create our stubs using the newStub and newBlockingStub methods provided in the RouteGuideGrpc class we generated from our .proto.

public RouteGuideClient(Channel channel) {
    blockingStub = RouteGuideGrpc.newBlockingStub(channel);
    asyncStub = RouteGuideGrpc.newStub(channel);
  }

Remember, if it's not blocking, it's async

Call service methods

Now let's look at how we call our service methods. Note that any RPCs created from the blocking stub will 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

Next, let's look at a server-side streaming call to ListFeatures, which returns a stream of geographical Feature:

Rectangle request = Rectangle.newBuilder()
             .setLo(Point.newBuilder().setLatitude(lowLat).setLongitude(lowLon).build())
        .setHi(Point.newBuilder().setLatitude(hiLat).setLongitude(hiLon).build()).build();

Iterator<Feature> features;
try {
  features = blockingStub.listFeatures(request);
} catch (StatusRuntimeException e) {
  logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
  return;
}

As you can see, it's very similar to the simple unary RPC we looked at in the Getting_Started_With_gRPC_Java codelab, except instead of returning a single Feature, the method returns an Iterator that the client can use to read all the returned Features.

Client-side streaming RPC

Now for something a little more complicated: the client-side streaming method RecordRoute, where we send a stream of Points to the server and get back a single RouteSummary. For this method we need to use the asynchronous stub. If you've already read Creating the server, some of this may look very familiar - asynchronous streaming RPCs are implemented in a similar way on both sides.

public void recordRoute(List<Feature> features, int numPoints) throws InterruptedException {
  info("*** RecordRoute");
  final CountDownLatch finishLatch = new CountDownLatch(1);
  StreamObserver<RouteSummary> responseObserver = new StreamObserver<RouteSummary>() {

    @Override
    public void onNext(RouteSummary summary) {
      info("Finished trip with {0} points. Passed {1} features. "
          + "Travelled {2} meters. It took {3} seconds.", summary.getPointCount(),
          summary.getFeatureCount(), summary.getDistance(), summary.getElapsedTime());
    }

    @Override
    public void onError(Throwable t) {
      Status status = Status.fromThrowable(t);
      logger.log(Level.WARNING, "RecordRoute Failed: {0}", status);
      finishLatch.countDown();
    }

    @Override
    public void onCompleted() {
      info("Finished RecordRoute");
      finishLatch.countDown();
    }
  };

  StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);
  try {
    // Send numPoints points randomly selected from the features list.
    Random rand = new Random();
    for (int i = 0; i < numPoints; ++i) {
      int index = rand.nextInt(features.size());
      Point point = features.get(index).getLocation();
      info("Visiting point {0}, {1}", RouteGuideUtil.getLatitude(point),
          RouteGuideUtil.getLongitude(point));
      requestObserver.onNext(point);
      // Sleep for a bit before sending the next one.
      Thread.sleep(rand.nextInt(1000) + 500);
      if (finishLatch.getCount() == 0) {
        // RPC completed or errored before we finished sending.
        // Sending further requests won't error, but they will just be thrown away.
        return;
      }
    }
  } catch (RuntimeException e) {
    // Cancel RPC
    requestObserver.onError(e);
    throw e;
  }
  // Mark the end of requests
  requestObserver.onCompleted();

  // Receiving happens asynchronously
  finishLatch.await(1, TimeUnit.MINUTES);
}

As you can see, to call this method we need to create a StreamObserver, which implements a special interface for the server to call with its RouteSummary response. In our StreamObserver we:

  • Override the onNext() method to print out the returned information when the server writes a RouteSummary to the message stream.
  • Override the onCompleted() method (called when the server has completed the call on its side) to reduce a CountDownLatch so that we can check to see if the server has finished writing.

We then pass the StreamObserver to the asynchronous stub's recordRoute() method and get back our own StreamObserver request observer to write our Points to send to the server. Once we've finished writing points, we use the request observer's onCompleted() method to tell gRPC that we've finished writing on the client side. Once we're done, we check our CountDownLatch to see if the server has completed on its side.

Bidirectional streaming RPC

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

public CountDownLatch routeChat() {
    info("*** RouteChat");
    final CountDownLatch finishLatch = new CountDownLatch(1);
    StreamObserver<RouteNote> requestObserver =
        asyncStub.routeChat(new StreamObserver<RouteNote>() {
          @Override
          public void onNext(RouteNote note) {
            info("Got message \"{0}\" at {1}, {2}", note.getMessage(), note.getLocation()
                .getLatitude(), note.getLocation().getLongitude());
          }

          @Override
          public void onError(Throwable t) {
            warning("RouteChat Failed: {0}", Status.fromThrowable(t));
            finishLatch.countDown();
          }

          @Override
          public void onCompleted() {
            info("Finished RouteChat");
            finishLatch.countDown();
          }
        });

    try {
      RouteNote[] requests =
          {newNote("First message", 0, 0), newNote("Second message", 0, 10_000_000),
              newNote("Third message", 10_000_000, 0), newNote("Fourth message", 10_000_000, 10_000_000)};

      for (RouteNote request : requests) {
        info("Sending message \"{0}\" at {1}, {2}", request.getMessage(), request.getLocation()
            .getLatitude(), request.getLocation().getLongitude());
        requestObserver.onNext(request);
      }
    } catch (RuntimeException e) {
      // Cancel RPC
      requestObserver.onError(e);
      throw e;
    }
    // Mark the end of requests
    requestObserver.onCompleted();

    // return the latch while receiving happens asynchronously
    return finishLatch;
  }

As with our client-side streaming example, we both get and return a StreamObserver response observer, except this time we send values via our method's response observer while the server is still writing messages to their message stream. The syntax for reading and writing here is exactly the same as for our client-streaming method. 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!

  1. From the start_here directory:
$ ./gradlew installDist

This will compile your code, package it in a jar and create the scripts that run the example. They will be created in the build/install/start_here/bin/ directory. The scripts are: route-guide-server and route-guide-client.

The server needs to be running before starting the client.

  1. Run the server:
$ ./build/install/start_here/bin/route-guide-server
  1. Run the client:
$ ./build/install/start_here/bin/route-guide-client

8. What's next