gRPC Streaming — Server & Bidirectional Guide

DodaTech Updated 2026-06-24 5 min read

In this tutorial, you'll learn about gRPC streaming. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

gRPC streaming enables long-lived connections where servers push multiple responses, clients send multiple requests, or both sides exchange data asynchronously over a single HTTP/2 stream.

What You'll Learn

You will learn the three gRPC streaming patterns — server-side, client-side, and bidirectional — with Protocol Buffer definitions and Python implementations using a threat intelligence feed as the example.

Why Streaming Matters

Unary RPCs (request-response) do not scale for real-time data feeds. Streaming eliminates polling overhead and enables instant event propagation. DodaTech's Durga Antivirus Pro uses gRPC bidirectional streaming to push threat intelligence updates to enterprise firewalls in real time.

Defining Streaming Services in Protobuf

syntax = "proto3";

package dodatech.threat;

service ThreatFeed {
  // Unary: single request, single response
  rpc GetThreat(ThreatRequest) returns (Threat);

  // Server-side streaming: one request, stream of responses
  rpc ListThreats(ThreatFilter) returns (stream Threat);

  // Client-side streaming: stream of requests, one response
  rpc SubmitThreats(stream ThreatSubmission) returns (SubmissionSummary);

  // Bidirectional streaming: stream of requests, stream of responses
  rpc MonitorThreats(stream ThreatQuery) returns (stream ThreatAlert);
}

message ThreatRequest {
  string threat_id = 1;
}

message Threat {
  string id = 1;
  string name = 2;
  string severity = 3;
  string detected_at = 4;
  bytes hash = 5;
}

message ThreatFilter {
  string severity = 1;
  int32 limit = 2;
}

message ThreatSubmission {
  string name = 1;
  string hash = 2;
  string source = 3;
}

message SubmissionSummary {
  int32 accepted = 1;
  int32 rejected = 2;
  repeated string errors = 3;
}

message ThreatQuery {
  string severity_filter = 1;
  int32 poll_interval_ms = 2;
}

message ThreatAlert {
  string threat_id = 1;
  string alert_type = 2;
  string message = 3;
}

Server-Side Streaming

import grpc
import threat_pb2
import threat_pb2_grpc
from concurrent import futures
import time

class ThreatFeedServicer(threat_pb2_grpc.ThreatFeedServicer):
    def ListThreats(self, request, context):
        """Server-side streaming: yields threats matching the filter."""
        limit = request.limit or 10
        count = 0

        for threat in fetch_threats_from_db(severity=request.severity):
            if count >= limit:
                break

            yield threat_pb2.Threat(
                id=threat["id"],
                name=threat["name"],
                severity=threat["severity"],
                detected_at=threat["detected_at"],
            )
            count += 1

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    threat_pb2_grpc.add_ThreatFeedServicer_to_server(
        ThreatFeedServicer(), server
    )
    server.add_insecure_port("[::]:50051")
    server.start()
    server.wait_for_termination()

Expected client output:

# Client receives threats one by one as they stream in
Threat: id="thr_001" name="Emotet" severity="high"
Threat: id="thr_002" name="Mirai" severity="critical"
Threat: id="thr_003" name="AgentTesla" severity="medium"

Client-Side Streaming

class ThreatFeedServicer(threat_pb2_grpc.ThreatFeedServicer):
    def SubmitThreats(self, request_iterator, context):
        """Client-side streaming: accepts a stream of submissions."""
        accepted = 0
        rejected = 0
        errors = []

        for submission in request_iterator:
            if validate_hash(submission.hash):
                store_threat(submission.name, submission.hash, submission.source)
                accepted += 1
            else:
                rejected += 1
                errors.append(f"Invalid hash for: {submission.name}")

        return threat_pb2.SubmissionSummary(
            accepted=accepted,
            rejected=rejected,
            errors=errors,
        )

Expected client output:

# Client streams threats
submissions = [
    threat_pb2.ThreatSubmission(name="Trojan A", hash="abc123", source="honeypot"),
    threat_pb2.ThreatSubmission(name="Ransomware B", hash="def456", source="partner"),
]
summary = stub.SubmitThreats(iter(submissions))
print(f"Accepted: {summary.accepted}, Rejected: {summary.rejected}")
# Accepted: 2, Rejected: 0

Bidirectional Streaming

class ThreatFeedServicer(threat_pb2_grpc.ThreatFeedServicer):
    def MonitorThreats(self, request_iterator, context):
        """Bidirectional streaming: client sends queries, server pushes alerts."""

        def process_alerts():
            for alert in threat_alert_generator():
                yield threat_pb2.ThreatAlert(
                    threat_id=alert["id"],
                    alert_type=alert["type"],
                    message=alert["message"],
                )

        # Stream alerts and handle client queries concurrently
        for client_query in request_iterator:
            print(f"Client watching severity: {client_query.severity_filter}")
            # Start alert stream
            for alert in process_alerts():
                if context.is_active():
                    yield alert

Expected behavior:

# Client connects and sends initial query
→ Server starts streaming alerts matching the filter
→ Client can send new queries to change the filter
→ Server adjusts the alert stream dynamically
→ Both sides exchange messages concurrently

flowchart LR
    subgraph "gRPC Streaming Patterns"
        SS["Server Streaming\nRequest → Stream Response\n(Threat Feed)"]
        CS["Client Streaming\nStream Request → Response\n(Batch Submission)"]
        BS["Bidirectional\nStream Request → Stream Response\n(Live Monitoring)"]
    end

    C["Client"] -->|"HTTP/2 Stream"| S["gRPC Server"]

    subgraph "Use Cases"
        U1["Real-time threat feed"]
        U2["Batch file upload"]
        U3["Live dashboard updates"]
    end

    SS --> U1
    CS --> U2
    BS --> U3

    style SS fill:#dbeafe,stroke:#2563eb
    style CS fill:#bbf7d0,stroke:#16a34a
    style BS fill:#fef3c7,stroke:#d97706

Client Implementation (Python)

import grpc
import threat_pb2
import threat_pb2_grpc

def run():
    channel = grpc.insecure_channel("localhost:50051")
    stub = threat_pb2_grpc.ThreatFeedStub(channel)

    # Server-side streaming
    print("=== Server Streaming: List Threats ===")
    for threat in stub.ListThreats(
        threat_pb2.ThreatFilter(severity="critical", limit=5)
    ):
        print(f"Threat: {threat.name} ({threat.severity})")

    # Bidirectional streaming
    print("=== Bidirectional: Monitor Threats ===")
    def client_queries():
        yield threat_pb2.ThreatQuery(severity_filter="critical")
        time.sleep(30)
        yield threat_pb2.ThreatQuery(severity_filter="all")

    for alert in stub.MonitorThreats(client_queries()):
        print(f"ALERT: {alert.message}")

if __name__ == "__main__":
    run()

Expected output:

=== Server Streaming: List Threats ===
Threat: Log4j Exploit (critical)
Threat: Zero-Day RCE (critical)
Threat: Ransomware Variant (critical)

=== Bidirectional: Monitor Threats ===
ALERT: New critical threat detected: ID thr_042
ALERT: Threat signature updated: ID thr_018

Common Errors

1. Not Handling Context Cancellation

When a client disconnects, the context.is_active() check prevents the server from continuing to stream data to a dead connection.

2. Blocking the Event Loop

Stream Processing with blocking I/O (database queries) holds up the gRPC Thread pool. Use async gRPC or a dedicated Thread pool for I/O-bound work.

3. Message Size Exceeds Limits

gRPC has a default message size limit of 4MB. Large threat payloads may need grpc.max_send_message_length and grpc.max_receive_message_length configuration.

4. Missing Error Handling in Streams

An exception in a streaming handler terminates the entire stream. Wrap stream logic in try/except blocks and log errors before re-raising.

5. No Keepalive Pings

Idle streams may be dropped by intermediate proxies. Configure gRPC keepalive pings to maintain the connection.

Practice Questions

1. What HTTP/2 feature enables gRPC streaming?

HTTP/2 multiplexed streams allow multiple concurrent messages over a single TCP connection without head-of-line blocking.

2. How does bidirectional streaming differ from server-side streaming?

Server-side: client sends one request, server sends multiple responses. Bidirectional: both sides send multiple messages independently over the same stream.

3. What is the default gRPC max message size?

4MB. This can be increased with channel and server configuration options.

4. Challenge: Implement a gRPC health check service with server-side streaming that reports server status changes.

Use a WatchHealth RPC that streams health status (SERVING, NOT_SERVING) whenever the server State changes, and a unary Check RPC for immediate status.

Mini Project: Threat Alert Stream

Build a gRPC bidirectional streaming service for DodaTech's threat alert system. Enterprise firewall clients connect, subscribe with severity filters, and receive real-time threat alerts. Include keepalive, error handling, and reconnection logic.