Bidirectional RPC Streaming

The Bidirectional-Streaming RPC Model

In bidirectional streaming, once the connection is initiated by the client, the communication channel is fully duplex. The client can send a message, then the server can send one, then the client can send three more, and the server can send two back—all without waiting for the other side to finish. The two streams are independent.

This diagram perfectly illustrates the concept.

Now, let's build this.

Step 1: Defining the Bidirectional Service

As always, we start in our .proto file. To define a bidirectional-streaming RPC, you place the stream keyword before both the request and response types.

We will add a new ChatMessage message and a Chat RPC to our UserService.

Update myservice/generated/user_service.proto:

syntax = "proto3";

package myservice;

// ... (existing messages: User, UserStatus, CreateUserRequest, CreateUserResponse, GetUsersRequest, ListUsersResponse)

// New message for our chat service
message ChatMessage {
  string user_id = 1;
  string text = 2;
}

service UserService {
  // Unary RPC
  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse);

  // Server-streaming RPC
  rpc GetUsers(GetUsersRequest) returns (stream User);

  // Client-streaming RPC
  rpc BatchCreateUsers(stream CreateUserRequest) returns (ListUsersResponse);

  // Bidirectional-streaming RPC
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

After updating the file, remember to regenerate the Python stubs. This step is crucial for making the new RPC available in your Python code. From your project's root directory, run:

python -m grpc_tools.protoc -I=myservice/generated --python_out=myservice/generated --pyi_out=myservice/generated --grpc_python_out=myservice/generated myservice/generated/user_service.proto

Step 2: Implementing the Server for Two-Way Communication

Implementing a bidirectional stream on the server is the most interesting challenge we've faced so far. The Chat method will receive an async iterator for client requests (request_iterator) and a context object with a write() method for sending responses.

The core problem is: how can we simultaneously listen for incoming messages and send outgoing messages? If we simply loop over request_iterator, our code will be blocked waiting for the client and won't be able to send any messages of its own.

The solution in an asyncio environment is to decouple the reading and writing logic into two separate, concurrent coroutines using asyncio.create_task.

This excellent Stack Overflow answer demonstrates the pattern clearly.

To facilitate communication between our reader and writer tasks, we will use an asyncio.Queue. The reader task will put incoming messages into the queue, and the writer task will get messages from the queue and send them to the client.

Let's modify myservice/server.py.

Update server.py:

# myservice/server.py

import asyncio
import logging
import uuid

import grpc
from generated import user_service_pb2
from generated import user_service_pb2_grpc




# ... (_USERS list and other methods remain the same)

class UserService(user_service_pb2_grpc.UserServiceServicer):



    # ... (CreateUser, GetUsers, BatchCreateUsers methods are unchanged)

    async def Chat(
        self,
        request_iterator: grpc.aio.AsyncIterator[user_service_pb2.ChatMessage],
        context: grpc.aio.ServicerContext,
    ) -> None:
        """
        Implements the Bidirectional-Streaming RPC for a chat service.
        """
        logging.info(f"Chat stream started by peer: {context.peer()}")
        



        # An asyncio.Queue to bridge the reader and writer coroutines.
        queue = asyncio.Queue()

        async def read_requests():
            """Coroutine to read messages from the client."""
            try:
                async for request in request_iterator:
                    logging.info(f"SERVER received: '{request.text}' from {request.user_id}")



                    # In a real app, you'd process the request.
                    # Here, we just put it on the queue for the writer to handle.
                    await queue.put(request)
            except grpc.aio.AioRpcError:
                logging.warning("Chat stream closed by client.")
            finally:



                # Signal the writer that there are no more incoming messages.
                await queue.put(None)

        async def write_responses():
            """Coroutine to send messages to the client."""
            while True:



                # Wait for a message from the reader coroutine.
                request = await queue.get()
                if request is None:



                    # End of stream signaled by the reader.
                    logging.info("SERVER closing response stream.")
                    break
                
                response_text = f"ECHO: {request.text.upper()}"
                logging.info(f"SERVER sending: '{response_text}'")
                



                # Create and send the response message.
                response = user_service_pb2.ChatMessage(
                    user_id="server",
                    text=response_text
                )
                await context.write(response)
                queue.task_done()
            
            queue.task_done()




        # Create and run the reader and writer tasks concurrently.
        read_task = asyncio.create_task(read_requests())
        write_task = asyncio.create_task(write_responses())




        # Wait for both tasks to complete.
        await asyncio.gather(read_task, write_task)
        logging.info("Chat stream finished.")

async def serve() -> None:



    # ... (no changes to this function)

if __name__ == "__main__":



    # ... (no changes here)

Key Architecture:

1. Chat Method: This is the main entry point. It sets up an asyncio.Queue and creates two tasks.
2. read_requests Task: Its only job is to async for over the request_iterator, receive messages from the client, and put them on the queue.
3. write_responses Task: Its job is to wait for messages to appear on the queue, process them (in our case, creating an uppercase echo), and send them back to the client using context.write().
4. asyncio.gather: This ensures that the Chat method doesn't exit until both the reading and writing operations are complete.

Step 3: Implementing the Concurrent Client

The client faces the same concurrency challenge: it needs to send messages (e.g., from user input) and receive messages from the server at the same time.

When you call a bidirectional streaming stub method, you pass it an async iterator (for your requests) and it returns an async iterator (for the server's responses). We'll again use asyncio.create_task to manage two coroutines.

Update client.py:

# myservice/client.py

import asyncio
import logging

import grpc
from generated import user_service_pb2
from generated import user_service_pb2_grpc




# ... (call_create_user, call_get_users, call_batch_create_users are unchanged)

async def call_chat(stub: user_service_pb2_grpc.UserServiceStub):
    """Calls the Bidirectional-Streaming Chat RPC."""
    logging.info("--- Client will call Chat ---")
    



    # An async generator to send a few predefined messages to the server.
    async def request_generator():
        messages_to_send = ["hello", "how are you?", "gRPC is cool!", "bye"]
        for msg_text in messages_to_send:
            logging.info(f"CLIENT sending: '{msg_text}'")
            yield user_service_pb2.ChatMessage(user_id="client-1", text=msg_text)
            await asyncio.sleep(1) # Wait a bit between messages

    try:



        # The stub call returns an async iterator for server responses.
        response_iterator = stub.Chat(request_generator(), timeout=30)




        # Use 'async for' to concurrently receive messages from the server.
        async for response in response_iterator:
            logging.info(f"CLIENT received: '{response.text}' from {response.user_id}")

    except grpc.aio.AioRpcError as e:
        logging.error(f"Chat RPC failed: {e.code()} - {e.details()}")

    logging.info("--- Chat finished ---")


async def run() -> None:
    """Creates a client and makes RPC calls to the server."""
    async with grpc.aio.insecure_channel("localhost:50051") as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)
        



        # For this lesson, let's focus only on the chat call.
        # You can comment out the previous calls if you like.
        # await call_create_user(stub)
        # print("-" * 20)
        # await call_get_users(stub)
        # print("-" * 20)
        # await call_batch_create_users(stub)
        # print("-" * 20)
        await call_chat(stub)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(run())

How the Client Works:
The async gRPC library elegantly handles the client-side concurrency for us in this common case.

1. We define an async generator request_generator that will produce the messages we want to send.
2. We call stub.Chat(), passing our generator. This call immediately returns an response_iterator.
3. The gRPC library starts a background task that consumes our request_generator and sends the messages to the server.
4. Simultaneously, our main client code can async for over the response_iterator to process messages as they arrive from the server. The two streams are handled concurrently by the library.

Step 4: Running the Application

It's time to see real-time communication in action.

Terminal 1: Start the Server
In your myservice/ directory:

python server.py

The server will start and log that it's waiting for connections.

Terminal 2: Run the Client
In another terminal, also in myservice/:

python client.py

Observe the logs in both terminals. You will see an interleaved "conversation". The client sends a message, the server receives it, the server sends an echo, and the client receives the echo, all while the client continues to send its next messages.

Client Output:

INFO:root:--- Client will call Chat ---
INFO:root:CLIENT sending: 'hello'
INFO:root:CLIENT received: 'ECHO: HELLO' from server
INFO:root:CLIENT sending: 'how are you?'
INFO:root:CLIENT received: 'ECHO: HOW ARE YOU?' from server
INFO:root:CLIENT sending: 'gRPC is cool!'
INFO:root:CLIENT received: 'ECHO: GRPC IS COOL!' from server
INFO:root:CLIENT sending: 'bye'
INFO:root:CLIENT received: 'ECHO: BYE' from server
INFO:root:--- Chat finished ---

Server Output:

INFO:root:gRPC server started on [::]:50051
INFO:root:Chat stream started by peer: ipv6:[::1]:...
INFO:root:SERVER received: 'hello' from client-1
INFO:root:SERVER sending: 'ECHO: HELLO'
INFO:root:SERVER received: 'how are you?' from client-1
INFO:root:SERVER sending: 'ECHO: HOW ARE YOU?'
INFO:root:SERVER received: 'gRPC is cool!' from client-1
INFO:root:SERVER sending: 'ECHO: GRPC IS COOL!'
INFO:root:SERVER received: 'bye' from client-1
INFO:root:SERVER sending: 'ECHO: BYE'
INFO:root:Chat stream closed by client.
INFO:root:SERVER closing response stream.
INFO:root:Chat stream finished.

This output clearly demonstrates the power of bidirectional streaming for building responsive, interactive systems.

Conclusion

Congratulations! You have now successfully implemented all four gRPC communication patterns. Bidirectional streaming is the most flexible of all, and understanding how to manage its concurrent nature is a key skill for building advanced microservices.

Key Takeaways:

• Service Definition: A bidirectional RPC is defined with the stream keyword on both the request and response types in the .proto file.
• Concurrent I/O: The core challenge is handling simultaneous reading and writing.
• Server Pattern: Use asyncio.create_task to create separate coroutines for reading from the client's request iterator and writing to the client's response stream. An asyncio.Queue is a great pattern for passing data between these tasks.
• Client Pattern: The async gRPC client library simplifies concurrency. You provide an async generator for requests and receive an async iterator for responses, which can be processed concurrently.

Preview of the Next Lesson

Mastering the four communication patterns is a major milestone. However, in real-world distributed systems, things can and do go wrong. In our next lesson, we will shift focus from communication patterns to robustness by learning how to handle gRPC errors gracefully. You'll learn about gRPC status codes and how to provide meaningful error details to your clients.