27 Adding Interceptors to Streaming RPC

Streaming RPC (Remote Procedure Call) in gRPC introduces a new paradigm for modern service communication. Beyond being efficient and scalable, streaming RPC also opens the door to sending data in real time and bidirectionally, making it essential in many microservices architecture use cases.

In production scenarios, however, simply transmitting data isn’t enough. There is often a need to add extra layers such as logging, authentication, rate limiting, observability, and even metering. This is exactly where the concept of interceptors comes into play.

In this article, I’ll walk through how to add interceptors to Streaming RPC in gRPC with Go, complete with code examples, scenario simulations, and flow diagrams. Let’s start by building a basic understanding of interceptors first.

What Is an Interceptor?

An interceptor in gRPC is similar to middleware in an HTTP server. It provides a structured way to perform pre-processing or post-processing on RPC requests and responses—from validation and auditing to handling errors consistently.

Broadly speaking, there are two types of interceptors in gRPC:

• Unary Interceptor: for unary RPC (single request & single response).
• Stream Interceptor: for streaming RPC (client, server, and bidirectional streaming).

In this article, our focus is on the stream interceptor.

Illustration of the Interceptor Flow

To give you a clearer picture, here is a diagram of how a request is processed with a stream interceptor:

sequenceDiagram
    participant Client
    participant Interceptor
    participant ServerHandler
    Client->>Interceptor: Kirim Stream Request
    Interceptor->>ServerHandler: Pre process
    ServerHandler->>Interceptor: Handler Response
    Interceptor->>Client: Post process lalu Kirim Response

sequenceDiagram
    participant Client
    participant Interceptor
    participant ServerHandler
    Client->>Interceptor: Kirim Stream Request
    Interceptor->>ServerHandler: Pre process
    ServerHandler->>Interceptor: Handler Response
    Interceptor->>Client: Post process lalu Kirim Response

With this scheme, the interceptor can run additional logic before or after the main handler is executed.

Case Study: A Logging Interceptor on Streaming RPC

Use Case

Let’s say you have a gRPC API that uses server streaming. Every time a client makes a streaming request, we want to log all incoming and outgoing data—without changing the original service logic.

Example service: a streaming stock price portfolio.

Proto Definition:

1service PortfolioService {
2    rpc StreamPrices (PriceRequest) returns (stream PriceUpdate) {};
3}

Implementing the Interceptor in Golang

Let’s break down how to implement it:

1. Interceptor Structure

gRPC Go provides a ServerStream Interceptor with the following signature:

1type StreamServerInterceptor func(srv interface{},
2    ss grpc.ServerStream,
3    info *grpc.StreamServerInfo,
4    handler grpc.StreamHandler) error

• srv: the gRPC service implementation.
• ss: the stream interface.
• info: the method metadata.
• handler: the main RPC handler, which we typically call inside the interceptor.

2. Implementing the Logging Stream Interceptor

Here is a simple interceptor that logs when streaming starts, when it ends, and any errors that occur:

 1func LoggingStreamInterceptor(
 2    srv interface{},
 3    ss grpc.ServerStream,
 4    info *grpc.StreamServerInfo,
 5    handler grpc.StreamHandler) error {
 6
 7    log.Printf("==== Start streaming RPC: %s, IsClientStream=%v, IsServerStream=%v ====",
 8        info.FullMethod, info.IsClientStream, info.IsServerStream)
 9
10    err := handler(srv, ss) // run the main handler
11
12    if err != nil {
13        log.Printf(">>> Stream method %s error: %v", info.FullMethod, err)
14    } else {
15        log.Printf("---- Streaming %s done ----", info.FullMethod)
16    }
17    return err
18}

3. Registering the Interceptor on the Server

When creating the gRPC server, we attach the stream interceptor through its options:

1server := grpc.NewServer(
2    grpc.StreamInterceptor(LoggingStreamInterceptor),
3)

For multiple interceptors, use chaining from a third-party package such as grpc-middleware .

4. Injecting a Custom ServerStream

If you want to log the payload or monitor all streaming communication (data reads/writes), you need to wrap grpc.ServerStream so you can intercept every message that is sent or received.

 1type loggingServerStream struct {
 2    grpc.ServerStream
 3}
 4
 5func (l *loggingServerStream) SendMsg(m interface{}) error {
 6    log.Printf("[SendMsg] %v", m)
 7    return l.ServerStream.SendMsg(m)
 8}
 9
10func (l *loggingServerStream) RecvMsg(m interface{}) error {
11    err := l.ServerStream.RecvMsg(m)
12    log.Printf("[RecvMsg] %v", m)
13    return err
14}

Then, inside the interceptor, the default stream is wrapped with the logging version:

 1func LoggingStreamInterceptor(
 2    srv interface{},
 3    ss grpc.ServerStream,
 4    info *grpc.StreamServerInfo,
 5    handler grpc.StreamHandler) error {
 6
 7    wrapped := &loggingServerStream{ss}
 8
 9    return handler(srv, wrapped)
10}

With this pattern, every streaming message going out or coming in will be logged.

Usage Simulation

a. The Client Sends a Request

The client sends a request to stream stock prices:

 1// Open the stream
 2stream, _ := client.StreamPrices(ctx, &PriceRequest{Symbol: "BBCA"})
 3
 4for {
 5    update, err := stream.Recv()
 6    if err == io.EOF {
 7        break
 8    }
 9    log.Printf("Price update: %v", update)
10}

b. Server Logging Output

Server logs after the interceptor is active:

1==== Start streaming RPC: /PortfolioService/StreamPrices, IsClientStream=false, IsServerStream=true ====
2[SendMsg] &PriceUpdate{...}
3[SendMsg] &PriceUpdate{...}
4---- Streaming /PortfolioService/StreamPrices done ----

Unary vs. Stream Interceptor Comparison

Conclusion

Adding interceptors to Streaming RPC in gRPC gives software engineers extra power to manage cross-cutting concerns without polluting the core business logic. From logging and authentication to observability, everything can be injected via interceptors—either globally or per method.

The practical steps above can be taken further: you can add rate limiting, auditing, and even metadata-based security. It’s no surprise that interceptors have become a crucial foundation in modern gRPC architecture.

If you have other experiments with interceptors on streaming RPC, feel free to share them in the comments! Happy experimenting and happy coding! 🚀