The Complete Guide to gRPC Part 4: Streams

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

Streaming#

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

Serverside streaming: When the request is sent in a simple way (unary), but the server response is a data stream.
Clientside streaming: It is the opposite of the previous one, when we have the request being sent in the form of data streams and the server response is unary.
Duplex streaming: When both the request and response are data streams.

This is reflected in a .proto file in a very simple way. Let’s go back to our repository for the second article in 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 modify the call so that instead of sending a ready-made list of notes, we send a stream of notes as a response in 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 clientside 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 don’t yet know the concept of streams, don’t worry, I made a series of articles on iMasters just about that:

Basically, streams are a continuous flow of data that are loaded at the time of their reading. This model has several benefits, for example, when we are working with very large files or content, if we have to return this content to whoever asked for it, we would have to load the entire file into memory first, and then be able to respond.

If your file is, say, 3GB, then you’ll use 3GB of memory. Whereas in a stream, you’re showing the file as it is loaded and the content that came later is being discarded and freed from memory. This way you have much faster processing using far fewer resources.

In this talk I showed visually what this means:

For this reason, streams are widely used with large files and data because they can support a huge amount of information using very few resources.

Streams and gRPC#

Since it’s so simple to use streams in gRPC, it was to be expected that support for them in the protocol would be very good. And this is, in fact what happens, stream support in gRPC is one of the best that exists and integrates with almost all supported languages.

For this demonstration, we’ll use the same application that we used in article number 2, and we’ll make some changes to it to transform a unary call into an asynchronous call.

Let’s start from a base, we clone the original repository from article 2 so we can 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. And then we’ll remove NoteListResponse so we have a response only as Note, the file looks 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 is important to note that we are only removing the response entity because, as we are talking about a stream, obviously all the data that will come will be notes. If we kept it as a response type { note: { } }, with each chunk of the stream we would have a new note object that would have (of course), a note inside… This is quite repetitive.

Server#

The next step is to change our server, actually, only a small part of it. The first and simplest change we’ll make is to remove our small in-memory database that has 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 approximately 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
  }, { ...

Remember that 200 notes is not, by far, a large amount of data. This is just an example.

Now, we load the file at the top of our server with require (remember that this does not 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’ll change will be the definition of the List method. To do this let’s look at the old definition for a moment:

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

We have some things to change here:

The response can no longer be { notes }, because we’re no longer returning an object
We can’t return the entire file at once anymore, or our chunk will be too large, we’ll iterate line by line through notes to be able to return them to the client
The function signature no longer takes a callback

We’ll solve all of this the following way, first, instead of 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 write stream along with the call record, so, if we had any kind of parameter to be sent, we could obtain them through call.request.parametro.

Now let’s define that a chunk of our stream will be an individual note, so we’ll iterate through the array of notes and return the notes individually:

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, this is because we changed our response to be only a note and not an object with a note key.

It is also interesting to note that as soon as the call to write is detected, the response will be sent and the client will receive it, this is useful when we have to do some kind of processing, for example, if we needed to transform all titles to uppercase, we could do that transformation and send the results without waiting for all notes to be loaded.

Finally, we call call.end(), which is important, as it instructs the client to close the connection, if this is not done, the same client will not be able to make another call to the same service.

Client#

For the client, very little will change, in fact, only 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 no longer call it in two ways, because the stream is mandatory asynchronous. Furthermore, we won’t have a callback, instead we’ll make the call to the server which will return a read stream to us, and only after we create a listener for this stream, will the call actually be made and the data returned.

This means we’ll be working with the event emitter and event listener pattern, 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 this way:

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

The stream also has another event called end, which is executed when the server stream calls the call.end() method. To listen to it, just create another listener;

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

Clientside streaming#

To complete the article and not leave anything behind. In the case of using a model like:

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

Where the client making the request uses streams, we’ll have a similar implementation on the server. The main difference is that our Find method, on the server side, will receive, as the first parameter, the client stream and the second will continue to be the callback.

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

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

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

1
function Find (call, callback) { }

And we can receive data from the client the same way we receive data from the server in serverside streaming:

1
function Find (call, callback) {
2
    call.on('data', (data) => {
3
        // fazer algo
4
    })
5
    call.on('end', () => {
6
        // a chamada terminou
7
    })
8
}

And on the client, we’ll have a call exactly like the server, but we have to account for the fact that the server, this time, does not return us a stream, 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 internal function of find will only be executed after the end() method is called.

Duplex streams#

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

On the server we would have something like:

1
function duplex (call) {
2
    call.on('data', (data) => {
3
        // recebendo dados do cliente
4
    })
5
    call.write('devolvendo dados para o cliente')
6
    call.end() // servidor encerra a conexão
7
}

And on the client we would have a call like:

1
const duplex = client.duplex()
2
duplex.on('data' (data) => {
3
  // recebe dados do servidor
4
})
5
duplex.write('envia dados ao servidor')
6
duplex.close() // client fecha conexão

The code for this demonstration is on my GitHub ↩