33 Writing Your Own Stream Interceptor

If you have ever built a gRPC-based application, you are surely already familiar with the concept of interceptors. In the gRPC ecosystem, interceptors play a role much like middleware in web frameworks such as Express (Node.js) or Gin (Go). However, there is one fundamental difference: in addition to interceptors for unary RPC, gRPC has a dedicated mechanism for handling streaming RPC, namely the Stream Interceptor. This time, we are going to dissect how to write your own stream interceptor—from the concept and design all the way to implementation and a simulated usage scenario.

Why Are Stream Interceptors Important?

gRPC supports several types of RPC:

• Unary: One request, one response.
• Server streaming: One request, many responses.
• Client streaming: Many requests, one response.
• Bidirectional streaming: Many requests and many responses simultaneously.

When we deal with streaming—whether from the client side, the server side, or both—standard middleware is not enough. This is exactly where stream interceptors become important, because they can:

• Perform validation or manipulation on every message (not just on the initial request/response).
• Handle logging, monitoring, per-message rate limiting, context manipulation, and even error handling across the data flow.
• Open the door to a variety of features such as per-stream metrics, tracing, and so on.

Anatomy: How Do Stream Interceptors Work in gRPC Go?

Let’s take a look at its signature (the Go version):

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

The important arguments here are:

• srv: the service implementation.
• ss: the stream object (used to send/receive messages).
• info: stream information (RPC name, method, etc.).
• handler: the “next” function, similar to a middleware handler on the web.

An overview of the interceptor flow:

flowchart LR
    Client--Stream Msgs--> Server[gRPC Server]
    Server-->|Incoming Context| Interceptor
    Interceptor-->|Manipulate or Pass| Handler
    Handler-->|Responses| Interceptor
    Interceptor-->|Manipulate or Pass| Client

flowchart LR
    Client--Stream Msgs--> Server[gRPC Server]
    Server-->|Incoming Context| Interceptor
    Interceptor-->|Manipulate or Pass| Handler
    Handler-->|Responses| Interceptor
    Interceptor-->|Manipulate or Pass| Client

Case Study: Building a Logging Stream Interceptor

Let’s implement logging on every message in a server-side stream. We want to know when the client sends/receives a message, as well as the processing time for each message.

1. Wrapping ServerStream

In gRPC Go, grpc.ServerStream is the interface for a stream. To intercept messages, we need to wrap the RecvMsg and SendMsg methods.

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

2. Writing the Interceptor

Next, we set up an interceptor that uses wrappedStream:

 1func LoggingStreamInterceptor(
 2    srv interface{},
 3    ss grpc.ServerStream,
 4    info *grpc.StreamServerInfo,
 5    handler grpc.StreamHandler,
 6) error {
 7    log.Printf("[Interceptor] Stream started: %s", info.FullMethod)
 8    err := handler(srv, &wrappedStream{ss})
 9    if err != nil {
10        log.Printf("[Interceptor] Stream error: %v", err)
11    } else {
12        log.Printf("[Interceptor] Stream completed successfully")
13    }
14    return err
15}

3. Registering the Interceptor on the Server

When building the server:

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

4. Simulation: A Simple Service

For example, a streaming upload service:

1service FileService {
2    rpc Upload(stream FileChunk) returns (UploadStatus);
3}
4message FileChunk {
5    bytes data = 1;
6}
7message UploadStatus {
8    string message = 1;
9}

Handler:

 1func (s *fileServiceServer) Upload(stream pb.FileService_UploadServer) error {
 2    var total int
 3    for {
 4        chunk, err := stream.Recv()
 5        if err == io.EOF {
 6            return stream.SendAndClose(&pb.UploadStatus{Message: "Received"})
 7        }
 8        if err != nil {
 9            return err
10        }
11        total += len(chunk.Data)
12        // Processing chunk...
13    }
14}

When the client streams a file, every chunk will be “logged” by the interceptor.

5. Simulation Output

The real-time logs will appear in the server console.

Modifying the Interceptor: Adding a Rate Limiting Feature

Let’s extend our interceptor by implementing a rate limiter based on messages per second.

1. A Simple Rate Limiter

 1type rateLimitStream struct {
 2    grpc.ServerStream
 3    tokens chan struct{}
 4}
 5
 6func NewRateLimitStream(ss grpc.ServerStream, rate int) *rateLimitStream {
 7    rl := &rateLimitStream{
 8        ServerStream: ss,
 9        tokens:       make(chan struct{}, rate),
10    }
11    // refill the tokens every second
12    go func() {
13        ticker := time.NewTicker(time.Second)
14        defer ticker.Stop()
15        for {
16            <-ticker.C
17            for i := 0; i < rate; i++ {
18                select {
19                case rl.tokens <- struct{}{}:
20                default:
21                }
22            }
23        }
24    }()
25    return rl
26}
27
28func (w *rateLimitStream) RecvMsg(m interface{}) error {
29    <-w.tokens // will block if the quota is exhausted
30    return w.ServerStream.RecvMsg(m)
31}

2. Integrating It into the Interceptor

 1func RateLimitStreamInterceptor(rate int) grpc.StreamServerInterceptor {
 2    return func(
 3        srv interface{},
 4        ss grpc.ServerStream,
 5        info *grpc.StreamServerInfo,
 6        handler grpc.StreamHandler,
 7    ) error {
 8        limited := NewRateLimitStream(ss, rate)
 9        return handler(srv, limited)
10    }
11}

3. Using It on the Server

1grpcServer := grpc.NewServer(
2    grpc.StreamInterceptor(RateLimitStreamInterceptor(5)), // max 5 messages/second
3)

Summary: Stream Interceptors Are “Middleware on Steroids”

Building your own stream interceptor in gRPC is extremely powerful for observability, security, or traffic-shaping needs. You can build granular logging, tracing, rate limiting, custom authentication, quota enforcement, and many other use cases—all from a single point of change without ever touching your service handlers.

👇 Production tips:

• Don’t forget to handle context cancellation/errors.
• If you want multiple interceptors, use chaining or integrate with a third-party library (for example, go-grpc-middleware ).
• Be careful with logging—never log sensitive data.
• Measure performance; interceptor overhead can affect stream throughput.

When will you write your own stream interceptor? When you want full control over all the data streaming into and out of the server, this is how you do it. Happy experimenting!

References:

• gRPC Interceptor Overview
• Go gRPC Middleware
• gRPC Streaming RPCs

If you found this article helpful, please bookmark it and share it with your fellow developers. See you in the next experiment! 🚀