In the modern landscape of distributed systems, latency is the enemy. Whether you are building a collaborative document editor, a live trading dashboard, or an IoT telemetry system, standard request-response patterns often fall short. This is where gRPC’s bidirectional streaming shines, allowing for continuous, low-latency data flow. However, to make these streams production-ready, you must integrate powerful interceptors for authentication, logging, and error handling. In this post, we will explore how to implement bidirectional streaming and leverage interceptors in Go to create a robust real-time data synchronization engine.
Understanding Bidirectional Streaming
Unlike server-side or client-side streaming, bidirectional streaming allows both the client and server to send and receive messages independently and concurrently. This is ideal for scenarios where data flows in both directions, such as syncing changes between a client and a central server. In Go, this is handled via the grpc.ServerStream, which provides channels for reading and writing messages concurrently.
To demonstrate this, let’s define a simple protobuf service for syncing configuration updates. The SyncConfig RPC will accept a stream of ConfigUpdate messages from the client and return a stream of SyncStatus messages.
Implementing the Bidirectional Stream in Go
The core challenge with bidirectional streaming is managing the lifecycle of the connection and handling concurrent reads and writes. In Go, we can use sync.WaitGroup to ensure both directions are handled properly. Below is a practical implementation of the server-side handler.
func (s *server) SyncConfig(stream pb.ConfigService_SyncConfigServer) error {
var wg sync.WaitGroup
wg.Add(2)
// 1. Handle incoming messages from the client
go func() {
defer wg.Done()
for {
update, err := stream.Recv()
if err == io.EOF {
return
}
if err != nil {
log.Printf("Error receiving update: %v", err)
return
}
// Process the configuration update
s.processUpdate(update)
// Send acknowledgment back to the client
status := &pb.SyncStatus{
Status: pb.SyncStatus_ACCEPTED,
Timestamp: time.Now().Unix(),
}
if err := stream.Send(status); err != nil {
log.Printf("Error sending status: %v", err)
return
}
}
}()
// 2. Handle outbound messages (e.g., server-initiated broadcasts)
go func() {
defer wg.Done()
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
broadcast := &pb.SyncStatus{
Status: pb.SyncStatus_HEARTBEAT,
Message: "Server sync active",
}
if err := stream.Send(broadcast); err != nil {
return
}
case <-stream.Context().Done():
return
}
}
}()
// Wait for both goroutines to finish
wg.Wait()
return nil
}
This implementation highlights a critical pattern: separating the read loop from the write loop. By using a sync.WaitGroup, we ensure the server does not close the stream until both the processing of incoming updates and the sending of heartbeats are complete.
The Power of Interceptors
While the streaming logic handles the data flow, interceptors handle the cross-cutting concerns. In gRPC, interceptors are functions that wrap around the RPC handler, allowing you to inspect or modify the request and response before and after the main business logic executes. This is crucial for real-time systems where every millisecond counts, and every connection must be secured.
Let’s implement a simple logging interceptor that tracks the duration of bidirectional streams. This helps in monitoring the health of long-lived connections.
Implementing an Interceptor for Logging
Interceptors in Go are typically unary or stream interceptors. Since we are dealing with a streaming RPC, we need a StreamServerInterceptor.
func loggingInterceptor(ctx context.Context, req interface{}, info *grpc.StreamServerInfo, handler grpc.StreamHandler) (interface{}, error) {
start := time.Now()
log.Printf("Started %s stream", info.FullMethod)
// Call the next handler in the chain
stream, err := handler(ctx, req)
if err != nil {
log.Printf("Failed %s stream: %v", info.FullMethod, err)
return nil, err
}
// Wrap the stream to log when it closes
wrappedStream := &loggingStreamWrapper{
Stream: stream.(grpc.ServerStream),
method: info.FullMethod,
start: start,
}
return nil, wrappedStream.SendAndClose(nil)
}
type loggingStreamWrapper struct {
grpc.ServerStream
method string
start time.Time
}
func (w *loggingStreamWrapper) SendMsg(m interface{}) error {
err := w.ServerStream.SendMsg(m)
if err != nil {
log.Printf("Error sending message in %s: %v", w.method, err)
}
return err
}
func (w *loggingStreamWrapper) RecvMsg(m interface{}) error {
err := w.ServerStream.RecvMsg(m)
if err != nil && err != io.EOF {
log.Printf("Error receiving message in %s: %v", w.method, err)
}
return err
}
Note that the above is a simplified example. In production, you would want to wrap the stream more comprehensively to capture metrics accurately. However, this illustrates how you can inject logic into the stream lifecycle.
Integrating Streams and Interceptors
To bring it all together, you register the interceptor when creating the gRPC server instance. The interceptor will automatically wrap your SyncConfig method, providing visibility into the performance and status of all bidirectional connections without cluttering your business logic.
opts := []grpc.ServerOption{
grpc.UnaryInterceptor(loggingInterceptor), // Note: For streaming, use grpc.StreamInterceptor
grpc.StreamInterceptor(loggingInterceptor),
}
s := grpc.NewServer(opts...)
pb.RegisterConfigServiceServer(s, &server{})
Conclusion
Implementing bidirectional streaming in Go gRPC unlocks the potential for truly real-time applications. By separating the read and write loops and managing concurrency with standard Go primitives like sync.WaitGroup, you can build reliable, high-throughput data pipelines. Furthermore, integrating interceptors ensures that these complex streams are observable, secure, and maintainable. As you move toward building more sophisticated real-time systems, mastering these patterns will be essential for delivering a seamless user experience.