gRPC Server Streaming — Real-Time Data Push to Clients
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.
gRPC server streaming allows a server to send a sequence of messages in response to a single client request, enabling real-time data push without client polling or multiple requests.
What You'll Learn
By the end of this lesson you will define server streaming RPCs in .proto files, implement streaming servers in Python, consume streams on the client side, handle stream errors and cancellation, and choose between streaming and polling.
Why It Matters
Many Microservices need to push data continuously: progress updates for long-running tasks, live financial data feeds, log streaming, or real-time notifications. Server streaming eliminates the need for polling, reducing network overhead and latency.
Real-World Use
DodaZIP's backup service uses gRPC server streaming to send real-time progress updates during large file backups. The client initiates a backup request and receives a stream of Progress messages showing files completed, bytes transferred, and estimated time remaining.
flowchart LR
A[Client] -->|Single Request| B[gRPC Server]
B -->|Response 1: Progress 25%| A
B -->|Response 2: Progress 50%| A
B -->|Response 3: Progress 75%| A
B -->|Response 4: Complete| A
style B fill:#2d3748,color:#fff
Defining Server Streaming
The .proto definition for a streaming response.
# streaming_proto.py
# Server streaming proto definition
def streaming_proto():
print("Server Streaming Proto Definition")
print("=" * 40)
print()
print('syntax = "proto3";')
print()
print("package backup.v1;")
print()
print('service BackupService {')
print(" rpc BackupFiles (BackupRequest) returns (stream BackupProgress);")
print(" rpc MonitorLogs (LogRequest) returns (stream LogEntry);")
print("}")
print()
print("message BackupRequest {")
print(" repeated string file_paths = 1;")
print(" string destination = 2;")
print("}")
print()
print("message BackupProgress {")
print(" string current_file = 1;")
print(" int32 files_completed = 2;")
print(" int32 total_files = 3;")
print(" int64 bytes_transferred = 4;")
print(" string status = 5;")
print("}")
print()
print("message LogRequest {")
print(" string service_name = 1;")
print(" string log_level = 2;")
print("}")
print()
print("message LogEntry {")
print(" string timestamp = 1;")
print(" string level = 2;")
print(" string message = 3;")
print("}")
print()
print("Keyword 'stream' before response type indicates server streaming")
streaming_proto()
Implementing the Streaming Server
Python server that streams responses.
# streaming_server.py
# Server streaming implementation
def streaming_server():
print("gRPC Server Streaming Implementation")
print("=" * 40)
print()
server_code = """
import grpc
import time
from concurrent import futures
import backup_pb2
import backup_pb2_grpc
class BackupServicer(backup_pb2_grpc.BackupServiceServicer):
def BackupFiles(self, request, context):
total = len(request.file_paths)
for i, file_path in enumerate(request.file_paths):
# Simulate backup work
time.sleep(0.5)
# Check if client cancelled
if context.is_active() is False:
break
# Send progress update
yield backup_pb2.BackupProgress(
current_file=file_path,
files_completed=i + 1,
total_files=total,
bytes_transferred=(i + 1) * 1024,
status="in_progress"
)
# Send completion
yield backup_pb2.BackupProgress(
current_file="",
files_completed=total,
total_files=total,
bytes_transferred=total * 1024,
status="completed"
)
def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
backup_pb2_grpc.add_BackupServiceServicer_to_server(
BackupServicer(), server
)
server.add_insecure_port('[::]:50052')
server.start()
server.wait_for_termination()
"""
print(server_code)
print()
print("Key points:")
print("- Use 'yield' instead of 'return' for streaming methods")
print("- Check context.is_active() to detect client cancellation")
print("- Each yield sends one message to the client")
streaming_server()
Implementing the Streaming Client
Consuming a stream on the client side.
# streaming_client.py
# Client consuming server stream
def streaming_client():
print("gRPC Streaming Client Implementation")
print("=" * 40)
print()
client_code = """
import grpc
import backup_pb2
import backup_pb2_grpc
def monitor_backup():
channel = grpc.insecure_channel('backup-service:50052')
stub = backup_pb2_grpc.BackupServiceStub(channel)
files = ["/data/doc1.pdf", "/data/doc2.pdf", "/data/photo.jpg"]
request = backup_pb2.BackupRequest(
file_paths=files,
destination="s3://backups/"
)
print("Starting backup...")
print()
# Iterate over the response stream
for progress in stub.BackupFiles(request, timeout=30):
print(f"File: {progress.current_file or '(finalizing)'}")
print(f"Progress: {progress.files_completed}/{progress.total_files}")
print(f"Transferred: {progress.bytes_transferred} bytes")
print(f"Status: {progress.status}")
print()
print("Backup completed!")
if __name__ == '__main__':
monitor_backup()
"""
print(client_code)
print()
print("Expected output:")
print(" Starting backup...")
print(" File: /data/doc1.pdf")
print(" Progress: 1/3")
print(" Status: in_progress")
print(" ... (continues for each file)")
streaming_client()
Streaming vs Polling
Comparison of streaming and polling approaches.
# streaming_vs_polling.py
# Comparison of streaming vs polling
def compare_streaming_polling():
print("Server Streaming vs Polling Comparison")
print("=" * 45)
print()
comparisons = [
("Latency", "Real-time", "Polling interval delay"),
("Network Overhead", "One connection, many messages", "Many HTTP requests"),
("Server Load", "Push-based, efficient", "Higher due to repeated requests"),
("Client Complexity", "Stream handler needed", "Simple loop with timer"),
("Firewall Compatibility", "May need config", "Standard HTTP works everywhere"),
("Bidirectional", "Yes (with bidi streaming)", "No (request-response only)"),
("Backpressure", "Built-in via flow control", "Managed by polling frequency"),
]
print(f"{'Aspect':25s} {'Streaming':35s} {'Polling':35s}")
print("-" * 95)
for aspect, streaming, polling in comparisons:
print(f"{aspect:25s} {streaming:35s} {polling:35s}")
compare_streaming_polling()
Common Mistakes
Forgetting to check context.is_active(): When the client cancels, the server should stop yielding. Without this check, the server wastes resources sending messages that will be discarded.
Blocking the stream with slow operations: If you need to do CPU-heavy work between stream messages, offload it to a separate thread pool. The stream should yield quickly.
Not setting timeouts: Streaming calls can also hang indefinitely. Set a reasonable timeout for the entire stream.
Ignoring backpressure: gRPC handles flow control automatically, but if the server produces messages faster than the client consumes them, memory usage grows on both sides.
Using streaming for single responses: If you always send exactly one response, use unary RPC instead. Streaming adds unnecessary complexity.
Practice Questions
How do you indicate a method is server streaming in .proto? Add the
streamkeyword before the return type:returns (stream Response).What Python keyword do you use in a streaming server to send messages?
yield. Each yielded message is sent to the client as part of the stream.How does the client detect the end of a server stream? The for loop over the stub call exits naturally when the server's generator is exhausted.
What is the advantage of server streaming over polling? Real-time data delivery with lower latency and network overhead compared to repeated HTTP requests.
Challenge: Implement a gRPC server streaming service for a real-time stock price monitor. The client subscribes with a list of ticker symbols and receives a stream of price updates. Handle client disconnect gracefully.
FAQ
Mini Project
Build a gRPC server streaming service for a system monitoring dashboard. The client subscribes with a MonitorRequest specifying which metrics to track (CPU, memory, disk). The server streams MetricSnapshot messages every second. Implement both server and client.
def monitor_service():
print("System Monitor gRPC Service")
print("=" * 40)
print()
print('syntax = "proto3";')
print()
print("package monitor.v1;")
print()
print('service MonitorService {')
print(" rpc Subscribe (MonitorRequest) returns (stream MetricSnapshot);")
print("}")
print()
print("message MonitorRequest {")
print(" repeated string metrics = 1;")
print(" int32 interval_ms = 2;")
print("}")
print()
print("message MetricSnapshot {")
print(" string timestamp = 1;")
print(" map<string, double> values = 2;")
print("}")
print()
print("Client sends metrics=['cpu', 'memory'], interval_ms=1000")
print("Server streams snapshots every second")
monitor_service()
What's Next
Next: gRPC Bidirectional Streaming for two-way streaming communication.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro