The Complete gRPC Guide Part 4: Streams

In the previous articles of this series we learned what gRPC is, how it works, and how we can use this protocol to move data between systems built with different technologies and languages. But all of that was done using only the simplest protobuf definition models, meaning we were sending a simple request and getting back a simple response in a client/server model.

Streaming#

Besides what’s called Unary Calls, we also have Streaming calls, which are nothing more than requests and responses made through an asynchronous data stream. There are three types of streaming calls in gRPC:

• Server-side streaming: the request is sent in a simple (unary) way, but the server’s response is a data stream.
• Client-side streaming: the opposite of the previous one, when the request is sent as a data stream and the server’s response is unary.
• Duplex streaming: when both the request and the response are data streams.

This is reflected inside a .proto file in a really simple way. Let’s go back to our repository for the second article of the series, where we have the following notes.proto file:

1
syntax = "proto3";
2

3
service NoteService {
4
  rpc List (Void) returns (NoteListResponse);
5
  rpc Find (NoteFindRequest) returns (NoteFindResponse);
6
}
7

8
// Entities
9
message Note {
10
  int32 id = 1;
11
  string title = 2;
12
  string description = 3;
13
}
14

15
message Void {}
16

17
// Requests
18
message NoteFindRequest {
19
  int32 id = 1;
20
}
21

22
// Responses
23
message NoteFindResponse {
24
  Note note = 1;
25
}
26

27
message NoteListResponse {
28
  repeated Note notes = 1;
29
}

If we wanted to change the call so that, instead of sending a ready-made list of notes, we sent a stream of notes as the response of the List service, we can simply add the word stream in the direction we want:

1
service NoteService {
2
  rpc List (Void) returns (stream NoteListResponse);
3
  rpc Find (NoteFindRequest) returns (NoteFindResponse);
4
}

Done! We don’t need to do anything else, our response will be a stream of notes as defined in NoteListResponse.

For the other stream models we can follow the same idea. If we want a client-side stream, we put stream only on the request side:

1
service NoteService {
2
  rpc List (Void) returns (NoteListResponse);
3
  rpc Find (stream NoteFindRequest) returns (NoteFindResponse);
4
}

And for duplex streams, we put stream on both sides:

1
service NoteService {
2
  rpc List (Void) returns (stream NoteListResponse);
3
  rpc Find (stream NoteFindRequest) returns (stream NoteFindResponse);
4
}

What are streams#

If you’re not familiar with the concept of streams yet, don’t worry, I wrote a whole series of articles on iMasters just about it:

• What are streams - part 1
• What are streams - part 2
• What are streams - part 3

Basically, streams are a continuous flow of data that’s loaded at the moment it’s read. This model has several benefits: for example, when we’re working with files or content that’s very large, if we had to return that content to whoever requested it, we’d have to load the entire file into memory first before we could respond.

If your file is, say, 3GB, then you’re going to use 3GB of memory. With a stream, on the other hand, you show the file as it’s loaded, and the content that already went by gets discarded and freed from memory. That way you get much faster processing while using far fewer resources.

In this talk I showed visually what this means:

That’s why streams are widely used with files and large-scale data: they can handle an enormous amount of information while using very few resources.

Streams and gRPC#

Since streams are so simple to use in gRPC, you’d expect the protocol’s support for them to be really good. And that’s exactly what happens: gRPC’s stream support is one of the best out there, and it integrates with almost every supported language.

For this demo, we’re going to use the same application we used in article number 2, and we’ll make a few changes to it to turn a unary call into an asynchronous call.The code for this demo is on my GitHub

Let’s start from a base: we clone the original repository from article 2 so we have the complete application. The first thing we need to do is change our .proto file to add a stream to the note listing service.

The first change is simply adding stream to rpc List. Then we’ll remove NoteListResponse so that our response is just Note. The file ends up like this:

1
syntax = "proto3";
2

3
service NoteService {
4
  rpc List (Void) returns (stream Note);
5
  rpc Find (NoteFindRequest) returns (NoteFindResponse);
6
}
7

8
// Entities
9
message Note {
10
  int32 id = 1;
11
  string title = 2;
12
  string description = 3;
13
}
14

15
message Void {}
16

17
// Requests
18
message NoteFindRequest {
19
  int32 id = 1;
20
}
21

22
// Responses
23
message NoteFindResponse {
24
  Note note = 1;
25
}

It’s worth pointing out that we’re only removing the response entity because, since we’re talking about a stream, obviously every piece of data that comes through will be a note. If we kept a response shaped like { note: { } }, every chunk of the stream would have a new note object which would, of course, have a note inside it… that’s pretty repetitive.

Server#

The next step is to change our server, or really, just a small part of it. The first and simplest change we’re going to make is to remove our little in-place database where we had our three fixed notes, and move it to a notes.json file that will represent a large amount of data.

In this file I put around 200 notes:

1
[
2
  {
3
    "id": 0,
4
    "title": "Note by Lucas Houston",
5
    "description": "Content http://hoateluh.md/caahaese"
6
  }, {
7
    "id": 1,
8
    "title": "Note by Brandon Tran",
9
    "description": "Content http://ki.bo/kuwokal"
10
  }, {
11
    "id": 2,
12
    "title": "Note by Michael Gonzalez",
13
    "description": "Content http://hifuhi.edu/cowkucgan"
14
  }, { ...

Keep in mind that 200 notes isn’t, by any stretch, a large amount of data. This is just an example.

Now we load the file at the top of our server with require (keep in mind this doesn’t work for ES Modules):

1
const grpc = require('grpc')
2
const protoLoader = require('@grpc/proto-loader')
3
const path = require('path')
4
const notes = require('../notes.json')

The second part of the file we’re going to change is the definition of the List method. Let’s take a look at the old definition for a moment:

1
function List (_, callback) {
2
  return callback(null, { notes })
3
}

There are a few things we need to change here:

1. The response can no longer be { notes }, because we’re no longer returning an object
2. We can no longer return the whole file at once, or our chunk would be way too big, so we’ll iterate note by note to return them to the client
3. The function signature no longer takes a callback

We’ll sort all of this out as follows. First, instead of the two parameters of a unary call, a stream only takes a single parameter, which we’ll call call:

1
function List (call) {
2
    //
3
}

The call object is an implementation of a writable stream together with the call’s metadata, so if we had some kind of parameter to send, we could get it through call.request.parametro.

Now let’s define that a chunk of our stream will be a single note, so we’ll iterate over the notes array and return the notes one by one:

1
function List (call) {
2
  for (const note of notes) {
3
    call.write(note)
4
  }
5
  call.end()
6
}

Notice that we’re calling call.write and passing the note directly, because we changed our response to be just a note and not an object with a note key.

It’s also worth noting that as soon as the call to write happens, the response is sent and the client receives it right away. This is useful when we need to do some kind of processing: for example, if we needed to uppercase all the titles, we could do that transformation and send the results out without waiting for every note to be loaded.

At the end, we call call.end(), which matters because it tells the client to close the connection. If we don’t do this, that same client won’t be able to make another call to the same service.

Client#

On the client side, very little is going to change, really just the method call. Our old call could be made in two ways:

1
client.listAsync({}).then(console.log)
2
client.list({}, (err, notes) => {
3
  if (err) throw err
4
  console.log(notes)
5
})

Now we can’t call it in two ways anymore, because a stream is necessarily asynchronous. On top of that, we won’t have a callback: instead, we make the call to the server, which gives us back a readable stream, and only after we set up a listener for that stream will the call actually happen and the data come back.

This means we’ll be working with the event emitter and event listener pattern, which is very common in Node and JavaScript. Our function will look like this:

1
const noteStream = client.list({})
2
noteStream.on('data', console.log)

To be more explicit, we can do it like this:

1
const noteStream = client.list({})
2
noteStream.on('data', (note) => console.log(note))

The stream also has another event called end, which fires when the server stream calls the call.end() method. To listen for it, we just create another listener:

1
noteStream.on('end', () => {})

Client-side streaming#

To wrap up the article and not leave anything out, in case we use a model like:

1
rpc Find (stream NoteFindRequest) returns (NoteFindResponse);

Where the client makes the request using streams, we’ll have a similar implementation on the server. The big difference is that our Find method, on the server side, will receive the client’s stream as its first parameter, and the second parameter will still be the callback.

This is our old method, with the two unary calls:

1
function Find ({ request: { id } }, callback) { }

It’s still valid because the call has a request property. But we don’t have the on method, so let’s update it to:

1
function Find (call, callback) { }

And we can receive the client’s data the same way we received the server’s data in server-side streaming:

1
function Find (call, callback) {
2
    call.on('data', (data) => {
3
        // do something
4
    })
5
    call.on('end', () => {
6
        // the call has ended
7
    })
8
}

And on the client, we’ll have a call that looks exactly like the server’s, but this time we have to account for the fact that the server doesn’t return a stream to us, so we have a callback:

1
const call = client.find((err, response) => {
2
    if (err) throw err
3
    console.log(response)
4
})
5

6
call.write({ id: 1 })
7
call.end()

The inner function of find will only run after the end() method is called.

Duplex streams#

For duplex streams (or bidirectional streams), all we need to do, on both the server and client side, is implement the call parameter. This parameter is a bidirectional stream that has both the on method and the write method.

On the server we’d have something like:

1
function duplex (call) {
2
    call.on('data', (data) => {
3
        // receiving data from the client
4
    })
5
    call.write('sending data back to the client')
6
    call.end() // server closes the connection
7
}

And on the client we’d have a call like:

1
const duplex = client.duplex()
2
duplex.on('data' (data) => {
3
  // receives data from the server
4
})
5
duplex.write('sends data to the server')
6
duplex.close() // client closes connection