gRPC Server Streaming — Streaming Data from Server to Client

In this tutorial, you will learn about grpc server streaming. We cover key concepts, practical examples, and best practices to help you master this topic.

Server-side streaming RPCs send a stream of responses to a single client request, enabling efficient delivery of large datasets and real-time event feeds.

Server Streaming Definition

service ThreatFeed {
  rpc ListThreats(ListThreatsRequest) returns (stream Threat);
  rpc WatchAlerts(WatchAlertsRequest) returns (stream Alert);
}

Server Implementation

class ThreatFeedServicer(threat_pb2_grpc.ThreatFeedServicer):
    def ListThreats(self, request, context):
        limit = request.limit or 100
        threats = db.get_threats(severity=request.severity, limit=limit)
        
        for threat in threats:
            yield threat_pb2.Threat(
                id=threat.id,
                name=threat.name,
                severity=threat.severity,
                detected_at=threat.detected_at.isoformat(),
            )
    
    def WatchAlerts(self, request, context):
        """Stream alerts as they occur in real-time."""
        for alert in alert_generator():
            if not context.is_active():
                break
            yield threat_pb2.Alert(
                id=alert.id,
                message=alert.message,
                severity=alert.severity,
            )

Client Consumption

def run():
    channel = grpc.insecure_channel("localhost:50051")
    stub = threat_pb2_grpc.ThreatFeedStub(channel)
    
    # Consume server stream
    print("=== Critical Threats ===")
    for threat in stub.ListThreats(
        threat_pb2.ListThreatsRequest(severity="CRITICAL", limit=5)
    ):
        print(f"  {threat.name} ({threat.severity}) — {threat.detected_at}")
    
    # Watch alerts with timeout
    print("=== Watching Alerts (10 seconds) ===")
    try:
        for alert in stub.WatchAlerts(
            threat_pb2.WatchAlertsRequest(),
            timeout=10.0
        ):
            print(f"ALERT: {alert.message}")
    except grpc.RpcError as e:
        if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
            print("Watch completed (timeout)")

Common Mistakes

1. Loading All Data in Memory

Don't collect all results before starting to yield. Use database cursors or generators to stream results as they arrive.

2. Not Checking context.is_active()

If the client disconnects, continue yielding wasted data. Check context.is_active() in long-running streams.

3. Blocking the Event Loop

Yield operations should not block. Use async generators or offload to a separate thread for I/O-bound work.

4. Too Large Individual Messages

Each yielded message should be reasonably sized. For large payloads, paginate or chunk data.

5. No Error Handling in Streams

An exception in the generator terminates the stream. Wrap in try/except and log errors before re-raising.

Practice Questions

1. When should you use server streaming instead of unary?
2. How does the client consume a server stream?
3. Why check context.is_active() in a streaming handler?
4. What happens when an error occurs mid-stream?
5. How do you limit the number of items in a stream?

Answers:

1. When sending large datasets (1000+ items), real-time events, or progress updates. The client processes items as they arrive instead of waiting for all data.
2. Using a for loop over the stub method call: for item in stub.ListThreats(request):.
3. If the client disconnects, is_active() returns False. Continue yielding wastes resources and may cause errors.
4. The stream terminates and the error is sent to the client. Wrap stream logic in try/except for graceful handling.
5. Use a count variable in the generator and break after reaching the limit. The proto message can include a limit field.

Mini Project

Build a server streaming gRPC service for DodaTech's threat feed. Implement ListThreats (with severity filter and limit) and WatchAlerts (real-time streaming using a generator with context.is_active() checks).

What's Next