Introduction
gRPC streaming allows protobuf messages to be streamed from client to server, server to client, or bidirectionally.
This powerful feature can be used to build real-time applications such as chat applications, real-time monitoring dashboards, and more.
In this article, we will explore how to use gRPC streaming correctly.
Prerequisites
- Basic knowledge of gRPC
- Basic knowledge of Go programming language (The sample code is written in Go, but the concept can be applied to other languages as well)
- The code examples are available on
serverStreamwraps the stream and the sensor data to make it easier to work with.
CODEtype serverStream struct {
s *sensorService // Service
stream pb.Sensor_WatchServer // Stream
sendCh chan *pb.WatchResponse // Control channel
sensorCh chan sensorData // Data channel
sensorWatch map[string]int // Map of sensor id to watch id
}
As noted before, the server can send and receive messages at the same time, one
function will handle the incoming messages and another function will handle the
outgoing messages.
Receiving messages:
CODEfunc (ss *serverStream) recvLoop() error {
defer ss.close()
for {
req, err := ss.stream.Recv()
if errors.Is(err, io.EOF) {
return nil
}
if err != nil {
return err
}
switch req := req.Request.(type) {
case *pb.WatchRequest_CreateRequest:
// IGNORE VALIDATION (check the full code)
// create a channel to send data to the client
id := sensor.watch(ss.sensorCh)
ss.sensorWatch[sensorId] = id
// send created message
ss.sendCh <- &pb.WatchResponse{
SensorId: sensorId,
Created: true,
}
case *pb.WatchRequest_CancelRequest:
// IGNORE VALIDATION (check the full code)
// cancel the watch
ss.s.sensors[sensorId].cancel(id)
delete(ss.sensorWatch, sensorId)
ss.sendCh <- &pb.WatchResponse{
SensorId: sensorId,
Canceleted: true,
}
case *pb.WatchRequest_NowRequest:
// IGNORE VALIDATION (check the full code)
// send current value
ss.sendCh <- &pb.WatchResponse{
SensorId: sensorId,
Timestamp: timestamppb.Now(),
Value: int32(sensor.read()),
}
}
}
}
The switch statement is used to handle the different types of requests and decide
what to do with each request. It's important to leave the recvLoop function only
to read and don't send messages to the client for this reason we have the sendLoop
that will read the messages from the control channel and send it to the client.
Sending messages:
CODEfunc (ss *serverStream) sendLoop() {
for {
select {
case m, ok := <-ss.sendCh:
if !ok {
return
}
// send message
if err := ss.stream.Send(m); err != nil {
return
}
case data, ok := <-ss.sensorCh:
if !ok {
return
}
// send data
if err := ss.stream.Send(&pb.WatchResponse{
SensorId: data.id,
Timestamp: timestamppb.New(data.time),
Value: int32(data.val),
}); err != nil {
return
}
case <-ss.stream.Context().Done():
return
}
}
}
The sendLoop function reads both the control channel and the data channel and sends
the messages to the client. If the stream is closed, the function will return.
Finally, a happy path test for the sensor service:
CODEfunc TestSensor(t *testing.T) {
conn := newServer(t, func(s grpc.ServiceRegistrar) {
pb.RegisterSensorServer(s, &sensorService{
sensors: newSensors(),
})
})
client := pb.NewSensorClient(conn)
stream, err := client.Watch(context.Background())
if err != nil {
t.Fatalf("failed to watch: %v", err)
}
response := make(chan *pb.WatchResponse)
// Go routine to read from the stream
go func() {
defer close(response)
for {
resp, err := stream.Recv()
if errors.Is(err, io.EOF) {
return
}
if err != nil {
return
}
response <- resp
}
}()
createRequest(t, stream, "temp")
waitUntilCreated(t, response, "temp")
waitForSensorData(t, response, "temp")
createRequest(t, stream, "pres")
waitUntilCreated(t, response, "pres")
waitForSensorData(t, response, "pres")
waitForSensorData(t, response, "temp")
waitForSensorData(t, response, "pres")
// invalid sensor
createRequest(t, stream, "invalid")
waitUntilCanceled(t, response, "invalid")
nowRequest(t, stream, "light")
waitForSensorData(t, response, "light")
// Wait for 2 seconds to make sure we don't receive any data for light
waitForNoSensorData(t, response, "light", 2*time.Second)
cancelRequest(t, stream, "temp")
waitUntilCanceled(t, response, "temp")
waitForSensorData(t, response, "pres")
// Wait for 2 seconds to make sure we don't receive any data for temp
waitForNoSensorData(t, response, "temp", 2*time.Second)
err = stream.CloseSend()
if err != nil {
t.Fatalf("failed to close send: %v", err)
}
}
From the test above, we can see that the client can create, cancel, and get the current
value of a sensor. The client can also watch multiple sensors at the same time.
Challenge Yourself
- Implement a chat application using gRPC streaming.
- Modify the sensor service to send multiple values at once to save round trips.
- Sniff the network traffic to see the difference between unary request and streaming request.
Conclusion
gRPC streaming is a versatile and powerful tool for building real-time applications.
By following best practices like using streaming only when necessary, batching data efficiently, and leveraging bidirectional streaming wisely, developers can maximize performance
and maintain code simplicity.
While gRPC streaming introduces complexity, its benefits far outweigh the challenges
when applied thoughtfully.
Stay in touch
If you have any questions or feedback, feel free to reach out to me on LinkedIn.
↗ Original-Artikel auf dev.to lesenVollständiger Original-BerichtAusführliche Details, Code-Beispiele & Hersteller-Stellungnahme auf dev.to.
SOCIAL SHARE CARD GENERATOR