Day 7: Challenge — Bidirectional gRPC Chat Service

Phase 3 · Jul 7, 2026

Challenge Overview

Build a bidirectional streaming gRPC chat service with named rooms.

Requirements:

1. .proto with a Chat bidirectional streaming RPC
2. Server maintains rooms; clients join and send messages
3. Messages broadcast to all members of the same room
4. Client disconnect is detected and room membership cleaned up
5. Messages include sender ID, timestamp, and room ID
6. Test with two grpcurl sessions in parallel

Proto Definition

syntax = "proto3";
package chat.v1;
import "google/protobuf/timestamp.proto";

message ChatMessage {
  string room_id = 1;
  string sender_id = 2;
  string text = 3;
  google.protobuf.Timestamp sent_at = 4;
}

message JoinRequest {
  string room_id = 1;
  string user_id = 2;
}

service ChatService {
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

Server Architecture

struct ChatServer {
    rooms: Arc<DashMap<String, broadcast::Sender<ChatMessage>>>,
}

#[tonic::async_trait]
impl ChatService for ChatServer {
    type ChatStream = Pin<Box<dyn Stream<Item = Result<ChatMessage, Status>> + Send>>;

    async fn chat(
        &self,
        request: Request<Streaming<ChatMessage>>,
    ) -> Result<Response<Self::ChatStream>, Status> {
        let mut inbound = request.into_inner();
        // First message must contain room_id + sender_id (join handshake)
        let first = inbound.message().await?.ok_or(Status::cancelled("no join"))?;
        let room_id = first.room_id.clone();
        // Subscribe to room broadcast
        let mut rx = self.get_or_create_room(&room_id).subscribe();
        // Spawn task to forward inbound messages to broadcast
        // Return outbound stream from broadcast receiver
    }
}

Connection Lifecycle

// Inbound → broadcast
tokio::spawn(async move {
    while let Ok(Some(msg)) = inbound.message().await {
        tx.send(msg).ok(); // Ignore if no receivers
    }
    // Inbound stream closed → client disconnected
    eprintln!("client disconnected from room {}", room_id);
});

// Broadcast → outbound stream
let output_stream = async_stream::stream! {
    loop {
        match rx.recv().await {
            Ok(msg) => yield Ok(msg),
            Err(broadcast::error::RecvError::Lagged(n)) => {
                yield Err(Status::data_loss(format!("lagged {} messages", n)));
                break;
            }
            Err(broadcast::error::RecvError::Closed) => break,
        }
    }
};

Testing

# Terminal 1: Alice joins room "rust"
grpcurl -plaintext -d '{"room_id":"rust","sender_id":"alice","text":"hello"}' \
    -rpc-header 'content-type: application/grpc' \
    localhost:50051 chat.v1.ChatService/Chat

# Terminal 2: Bob joins same room
grpcurl -plaintext -d '{"room_id":"rust","sender_id":"bob","text":"hi alice"}' \
    localhost:50051 chat.v1.ChatService/Chat

Both terminals should receive each other's messages.

Week 1 Recap

Next week: custom binary protocols — when gRPC is too heavy.