Streaming RPCs
Send many messages on one call: server, client and bidirectional streams, and how each one ends.
rpc Watch(HealthCheckRequest) returns (stream HealthCheckResponse);A unary rpc carries one message each way. A stream carries a sequence of messages on a single call, and the .proto file marks one with the word stream. Where that word goes decides the kind.
The three kinds
- Server streaming,
streamon the response: one request, many responses. Fits a large result set you would rather not buffer, or a feed of changes. - Client streaming,
streamon the request: many requests, one response. Fits an upload, or a batch of events summarised at the end. - Bidirectional,
streamon both: a conversation where neither side waits its turn.
The generated types
Each kind hands your method a stream object. A server-streaming method receives a grpc.ServerStreamingServer[Res], whose one added method is Send(*Res) error; the caller gets a grpc.ServerStreamingClient[Res] with Recv() (*Res, error). Client streaming swaps those around and adds SendAndClose on the server and CloseAndRecv on the client. Code written before these generic types used per-method names such as Health_WatchServer; the plugin still generates those names as aliases for the generic types, so the long name and the short one are the same type and you will meet both.
Ending a stream
A streaming method ends the call by returning: nil for success, an error to fail it. Whoever is reading sees io.EOF from Recv when the other side finished cleanly, and any other error is a real failure. On a bidirectional stream, read in one goroutine and write in another, because the directions are independent.
Example
package main
import (
"context"
"errors"
"fmt"
"io"
"log"
"net"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
healthpb "google.golang.org/grpc/health/grpc_health_v1"
)
type watcher struct {
healthpb.UnimplementedHealthServer
}
// Watch is server-streaming: one request, many responses. Returning nil
// closes the stream cleanly, which the client sees as io.EOF.
func (watcher) Watch(req *healthpb.HealthCheckRequest, stream grpc.ServerStreamingServer[healthpb.HealthCheckResponse]) error {
if err := stream.Send(&healthpb.HealthCheckResponse{Status: healthpb.HealthCheckResponse_NOT_SERVING}); err != nil {
return err // the client hung up
}
return stream.Send(&healthpb.HealthCheckResponse{Status: healthpb.HealthCheckResponse_SERVING})
}
func main() {
lis, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
log.Fatal(err)
}
srv := grpc.NewServer()
healthpb.RegisterHealthServer(srv, watcher{})
go func() { _ = srv.Serve(lis) }()
defer srv.Stop()
conn, err := grpc.NewClient(lis.Addr().String(),
grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
log.Fatal(err)
}
defer conn.Close()
stream, err := healthpb.NewHealthClient(conn).Watch(context.Background(), &healthpb.HealthCheckRequest{})
if err != nil {
log.Fatal(err)
}
for {
res, err := stream.Recv()
if errors.Is(err, io.EOF) { // the server returned: no more messages
break
}
if err != nil {
log.Fatal(err)
}
fmt.Println("status:", res.GetStatus())
}
}
// Output:
// status: NOT_SERVING
// status: SERVINGWhen to use it
- Stream a long query result to the client row by row instead of building one huge response.
- Push live status changes to a dashboard over a single long-lived call.
- Accept a large file as a sequence of chunks and reply with one summary.
- Run a two-way protocol, such as a worker asking for tasks while reporting progress.
More examples
All four shapes side by side
The only difference between the four is where the word stream appears, and that single word changes the Go signature the plugin generates. Choose by data shape, not by taste: List streams because a result set can be large, Upload streams because a file arrives in pieces, Chat streams both ways because neither side leads. A unary call is simpler to retry, so do not stream just because you can.
service Feed {
// unary: one in, one out
rpc Get(GetRequest) returns (Item);
// server streaming: one in, many out
rpc List(ListRequest) returns (stream Item);
// client streaming: many in, one out
rpc Upload(stream Chunk) returns (UploadSummary);
// bidirectional: many in, many out, in any order
rpc Chat(stream Message) returns (stream Message);
}Receiving a client stream
The loop that reads until the client is done. io.EOF here is not an error; it is the client saying it has sent everything, and it is the signal to produce the single response with SendAndClose. Anything else from Recv ends the call. Use this shape for uploads and for bulk ingest, where sending one huge message would mean holding the whole thing in memory on both sides.
func (s *feedServer) Upload(stream grpc.ClientStreamingServer[feedv1.Chunk, feedv1.UploadSummary]) error {
var total int64
for {
chunk, err := stream.Recv()
if errors.Is(err, io.EOF) {
// The client finished sending. Reply once and end the call.
return stream.SendAndClose(&feedv1.UploadSummary{Bytes: total})
}
if err != nil {
return err // a real transport failure, or the call was cancelled
}
total += int64(len(chunk.GetData()))
}
}Both directions at once
A bidirectional stream deadlocks if you try to drive it from one goroutine and the server replies out of step with your sends. One goroutine per direction is the standard answer, and each Send or Recv then has a single owner, which is what the concurrency rule requires. CloseSend half-closes: the server starts seeing io.EOF on its Recv while still being able to send to you.
func chat(ctx context.Context, client feedv1.FeedClient, outgoing []string) error {
stream, err := client.Chat(ctx)
if err != nil {
return err
}
g, _ := errgroup.WithContext(ctx)
g.Go(func() error { // one goroutine owns Send
for _, text := range outgoing {
if err := stream.Send(&feedv1.Message{Text: text}); err != nil {
return err
}
}
return stream.CloseSend() // nothing more from this side
})
g.Go(func() error { // a second goroutine owns Recv
for {
msg, err := stream.Recv()
if errors.Is(err, io.EOF) {
return nil
}
if err != nil {
return err
}
fmt.Println("server said:", msg.GetText())
}
})
return g.Wait()
}
Discussion