gRPC Bidirectional Streaming — Two-Way Data Exchange
In this tutorial, you will learn about grpc bidirectional streaming. We cover key concepts, practical examples, and best practices to help you master this topic.
gRPC bidirectional streaming allows both client and server to send independent streams of messages simultaneously over a single HTTP/2 connection, enabling real-time two-way communication patterns.
What You'll Learn
By the end of this lesson you will define bidirectional streaming RPCs, implement a server that reads from a request stream while writing to a response stream, build a client that sends and receives concurrently, and apply bidi streaming to real scenarios.
Why It Matters
Bidirectional streaming is the most powerful gRPC pattern. It enables use cases impossible with unary or server-only streaming: real-time chat, collaborative document editing, game state synchronization, and streaming data processing pipelines where both sides produce and consume data concurrently.
Real-World Use
DodaZIP's live collaboration feature uses bidirectional gRPC streaming so multiple users can edit file metadata simultaneously. Each client streams keystrokes to the server, which broadcasts changes to other connected clients in real time.
flowchart LR
subgraph Client
A[Send Stream]
B[Receive Stream]
end
subgraph Server
C[Receive Stream]
D[Broadcast]
E[Send Stream]
end
A --> C
D --> E
E --> B
style C fill:#2d3748,color:#fff
style D fill:#2d3748,color:#fff
Defining Bidirectional Streaming
The .proto definition with both streams.
# bidi_proto.py
# Bidirectional streaming proto
def bidi_proto():
print("Bidirectional Streaming Proto Definition")
print("=" * 45)
print()
print('syntax = "proto3";')
print()
print("package chat.v1;")
print()
print('service ChatService {')
print(" rpc Chat (stream ChatMessage) returns (stream ChatMessage);")
print(" rpc Collaborate (stream EditEvent) returns (stream DocumentState);")
print("}")
print()
print("message ChatMessage {")
print(" string user_id = 1;")
print(" string room_id = 2;")
print(" string text = 3;")
print(" int64 timestamp = 4;")
print("}")
print()
print("message EditEvent {")
print(" string user_id = 1;")
print(" string doc_id = 2;")
print(" string operation = 3;")
print(" int32 position = 4;")
print(" string text = 5;")
print("}")
print()
print("message DocumentState {")
print(" string doc_id = 1;")
print(" string content = 2;")
print(" int32 version = 3;")
print(" repeated string active_users = 4;")
print("}")
bidi_proto()
Implementing the Bidi Server
Server handling bidirectional streams.
# bidi_server.py
# Bidirectional streaming server
def bidi_server():
print("Bidirectional Streaming Server")
print("=" * 40)
print()
server_code = """
import grpc
import asyncio
from concurrent import futures
import chat_pb2
import chat_pb2_grpc
class ChatServicer(chat_pb2_grpc.ChatServiceServicer):
def __init__(self):
self.rooms = {} # room_id -> list of response streams
def Chat(self, request_iterator, context):
# Get first message to determine room
first_msg = next(request_iterator)
room_id = first_msg.room_id
# Register this client in the room
if room_id not in self.rooms:
self.rooms[room_id] = []
# We'll use a queue to broadcast to this client
response_queue = asyncio.Queue()
self.rooms[room_id].append(response_queue)
def broadcast_to_room(msg, exclude_queue):
for q in self.rooms.get(room_id, []):
if q is not exclude_queue:
q.put_nowait(msg)
# Broadcast the first message
broadcast_to_room(first_msg, response_queue)
# Process incoming messages and send queued ones
async def handle_stream():
# Send queued responses
while True:
try:
# Check for client messages
for msg in request_iterator:
if context.is_active():
broadcast_to_room(msg, response_queue)
except Exception:
break
# Send broadcast messages to this client
try:
response = await asyncio.wait_for(
response_queue.get(), timeout=1.0
)
yield response
except asyncio.TimeoutError:
if not context.is_active():
break
return handle_stream()
"""
print(server_code)
print()
print("Key points:")
print("- request_iterator yields client messages")
print("- Server yields responses while reading requests")
print("- Broadcast pattern forwards messages to all clients in room")
bidi_server()
Implementing the Bidi Client
Client sending and receiving simultaneously.
# bidi_client.py
# Bidirectional streaming client
def bidi_client():
print("Bidirectional Streaming Client")
print("=" * 40)
print()
client_code = """
import grpc
import threading
import chat_pb2
import chat_pb2_grpc
def send_messages(stub, room_id, user_id):
"""Generator that produces messages from this user."""
messages = [
"Hello everyone!",
"How is the project going?",
"I just pushed the latest changes.",
]
for text in messages:
yield chat_pb2.ChatMessage(
user_id=user_id,
room_id=room_id,
text=text,
timestamp=int(time.time())
)
time.sleep(1)
def receive_messages(response_stream, user_id):
"""Print incoming messages."""
for msg in response_stream:
if msg.user_id != user_id:
print(f"[{msg.user_id}]: {msg.text}")
def run_chat():
channel = grpc.insecure_channel('chat-service:50053')
stub = chat_pb2_grpc.ChatServiceStub(channel)
room_id = "project-alpha"
user_id = "user-42"
# Start the bidirectional stream
response_stream = stub.Chat(
send_messages(stub, room_id, user_id)
)
# Receive in background thread
receiver = threading.Thread(
target=receive_messages,
args=(response_stream, user_id)
)
receiver.start()
receiver.join()
if __name__ == '__main__':
import time
run_chat()
"""
print(client_code)
print()
print("Expected output:")
print(" [user-43]: Hey team!")
print(" [user-44]: Ready for review")
bidi_client()
Use Case Pattern: Stream Processing Pipeline
Bidirectional streaming for data processing.
# stream_pipeline.py
# Bidirectional stream processing
def stream_pipeline():
print("Stream Processing Pipeline with Bidi Streaming")
print("=" * 45)
print()
pipeline_code = """
import grpc
import transformer_pb2
import transformer_pb2_grpc
# Client sends chunks, server transforms and returns chunks
def transform_pipeline():
channel = grpc.insecure_channel('transformer:50054')
stub = transformer_pb2_grpc.TransformerStub(channel)
def produce_chunks():
data_chunks = ["hello", " world", " from", " gRPC", " bidi!"]
for chunk in data_chunks:
yield transformer_pb2.TransformRequest(
input_chunk=chunk,
transformation="uppercase"
)
response_stream = stub.Transform(produce_chunks())
result = ""
for response in response_stream:
result += response.output_chunk
print(f"Received chunk: '{response.output_chunk}'")
print(f"Final result: '{result}'")
# Expected output:
# Received chunk: 'HELLO'
# Received chunk: ' WORLD'
# Received chunk: ' FROM'
# Received chunk: ' GRPC'
# Received chunk: ' BIDI!'
# Final result: 'HELLO WORLD FROM GRPC BIDI!'
"""
print(pipeline_code)
stream_pipeline()
Common Mistakes
Assuming ordered request/response pairs: Bidi streams do not guarantee that responses correspond to specific requests in order. Design your protocol to handle interleaved messages.
Blocking the event loop while reading: Both reading from request_Iterator and yielding responses must be non-blocking. Use async or threads to handle both concurrently.
Not handling client disconnect: When the client disconnects, the request_iterator raises an exception. Always wrap it in try/except to clean up server resources.
Memory leaks from accumulated streams: If clients connect and never disconnect, server memory grows. Implement idle timeouts and periodic cleanup of disconnected streams.
Using bidi streaming for simple request-response: Bidi streaming adds complexity. Only use it when both sides need to send multiple independent messages.
Practice Questions
What makes bidirectional streaming different from server streaming? Both client and server send streams of messages independently. Server streaming only sends from server to client.
How does the server read client messages in a bidi RPC? By iterating over the request_iterator parameter in the server method.
What is the broadcast pattern in chat applications? When the server receives a message from one client, it forwards that message to all other connected clients in the same room.
How do you handle concurrent send and receive in a bidi client? Use a background thread to receive messages while the main thread sends messages, or use async/await.
Challenge: Implement a bidirectional streaming service for a collaborative code editor. Define events for cursor position, text insertions, and deletions. Handle concurrent edits from multiple users with operational transform.
FAQ
Mini Project
Build a bidirectional streaming service for a real-time auction system. Clients place bids via the request stream. The server broadcasts current highest bid and auction status to all connected clients. Handle bid validation, outbid notifications, and auction end.
def auction_service():
print("Real-Time Auction gRPC Service")
print("=" * 40)
print()
print('syntax = "proto3";')
print()
print("package auction.v1;")
print()
print('service AuctionService {')
print(" rpc Auction (stream Bid) returns (stream AuctionEvent);")
print("}")
print()
print("message Bid {")
print(" string user_id = 1;")
print(" string item_id = 2;")
print(" double amount = 3;")
print("}")
print()
print("message AuctionEvent {")
print(" string item_id = 1;")
print(" double current_bid = 2;")
print(" string highest_bidder = 3;")
print(" int32 time_remaining_sec = 4;")
print(" string status = 5;")
print("}")
print()
print("Flow: Client sends Bid -> Server validates")
print(" Server broadcasts AuctionEvent to all clients")
auction_service()
What's Next
Next: gRPC Auth for securing gRPC communication.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro