use mecha10::prelude::*;
use serde::{Deserialize, Serialize};
use std::time::Duration;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct VideoRequest {
camera_id: u32,
fps: u32,
}
impl Message for VideoRequest {}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct VideoFrame {
frame_num: u64,
timestamp: u64,
data: Vec<u8>,
}
impl Message for VideoFrame {}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SensorData {
value: f32,
timestamp: u64,
}
impl Message for SensorData {}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ChatMessage {
sender: String,
text: String,
timestamp: u64,
}
impl Message for ChatMessage {}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ProgressUpdate {
completed: u64,
total: u64,
message: String,
}
impl Message for ProgressUpdate {}
pub async fn server_streaming_video_example() -> Result<()> {
info!("Example 1: Server Streaming - Video feed");
info!("\nServer side - Stream video frames:");
info!("ctx.stream_respond(\"/camera/stream\", |req: VideoRequest, sender| async move {{");
info!(" for frame_num in 0..100 {{");
info!(" let frame = VideoFrame {{");
info!(" frame_num,");
info!(" timestamp: now(),");
info!(" data: capture_frame(req.camera_id),");
info!(" }};");
info!(" sender.send(&frame).await?;");
info!(" tokio::time::sleep(Duration::from_millis(1000 / req.fps)).await;");
info!(" }}");
info!(" sender.end().await?;");
info!(" Ok(())");
info!("}}).await?;");
info!("\nClient side - Receive video stream:");
info!("let mut stream = ctx");
info!(" .stream_request(\"/camera/stream\", &VideoRequest {{ camera_id: 0, fps: 30 }})");
info!(" .await?;");
info!("");
info!("while let Some(frame) = stream.recv().await {{");
info!(" process_frame(&frame);");
info!("}}");
Ok(())
}
pub async fn server_streaming_telemetry_example() -> Result<()> {
info!("Example 2: Server Streaming - Continuous sensor telemetry");
info!("\nServer side - Stream sensor data:");
info!("ctx.stream_respond(\"/sensors/telemetry\", |_req, sender| async move {{");
info!(" loop {{");
info!(" let data = SensorData {{");
info!(" value: read_sensor(),");
info!(" timestamp: now(),");
info!(" }};");
info!(" sender.send(&data).await?;");
info!(" ");
info!(" // Send heartbeat every 10 readings");
info!(" if reading_count % 10 == 0 {{");
info!(" sender.heartbeat().await?;");
info!(" }}");
info!(" ");
info!(" tokio::time::sleep(Duration::from_millis(100)).await;");
info!(" }}");
info!("}}).await?;");
info!("\nClient side - Monitor telemetry:");
info!("let mut stream = ctx.stream_request(\"/sensors/telemetry\", &()).await?;");
info!("");
info!("while let Some(data) = stream.recv().await {{");
info!(" println!(\"Sensor value: {} at {{}}\", data.value, data.timestamp);");
info!("}}");
Ok(())
}
pub async fn server_streaming_progress_example() -> Result<()> {
info!("Example 3: Server Streaming - Long-running task with progress");
info!("\nServer side - Stream progress updates:");
info!("ctx.stream_respond(\"/tasks/process\", |req: TaskRequest, sender| async move {{");
info!(" let total_items = req.items.len();");
info!(" ");
info!(" for (i, item) in req.items.iter().enumerate() {{");
info!(" // Process item");
info!(" process_item(item).await?;");
info!(" ");
info!(" // Send progress update");
info!(" sender.send(&ProgressUpdate {{");
info!(" completed: i as u64 + 1,");
info!(" total: total_items as u64,");
info!(" message: format!(\"Processed item {{}}\", i + 1),");
info!(" }}).await?;");
info!(" }}");
info!(" ");
info!(" sender.end().await?;");
info!(" Ok(())");
info!("}}).await?;");
info!("\nClient side - Monitor progress:");
info!("let mut stream = ctx.stream_request(\"/tasks/process\", &task_req).await?;");
info!("");
info!("while let Some(update) = stream.recv().await {{");
info!(" let progress = (update.completed as f32 / update.total as f32) * 100.0;");
info!(" println!(\"Progress: {{:.1}}% - {{}}\", progress, update.message);");
info!("}}");
Ok(())
}
pub async fn bidirectional_chat_example() -> Result<()> {
info!("Example 4: Bidirectional Streaming - Interactive chat");
info!("\nBidirectional chat example:");
info!("let (mut tx, mut rx) = ctx.bidirectional_stream(\"/chat/session\").await?;");
info!("");
info!("// Spawn receiver task");
info!("tokio::spawn(async move {{");
info!(" while let Some(msg) = rx.recv().await {{");
info!(" println!(\"[{{}}] {{}}\", msg.sender, msg.text);");
info!(" }}");
info!("}});");
info!("");
info!("// Send messages");
info!("tx.send(&ChatMessage {{");
info!(" sender: \"Alice\".into(),");
info!(" text: \"Hello, Bob!\".into(),");
info!(" timestamp: now(),");
info!("}}).await?;");
info!("");
info!("tx.send(&ChatMessage {{");
info!(" sender: \"Alice\".into(),");
info!(" text: \"How are you?\".into(),");
info!(" timestamp: now(),");
info!("}}).await?;");
info!("");
info!("// End conversation");
info!("tx.end().await?;");
Ok(())
}
pub async fn bidirectional_collaboration_example() -> Result<()> {
info!("Example 5: Bidirectional Streaming - Real-time document collaboration");
info!("\nCollaboration example:");
info!("let (mut tx, mut rx) = ctx.bidirectional_stream(\"/collab/doc\").await?;");
info!("");
info!("// Listen for changes from other users");
info!("tokio::spawn(async move {{");
info!(" while let Some(change) = rx.recv().await {{");
info!(" apply_change_to_document(&change);");
info!(" update_ui();");
info!(" }}");
info!("}});");
info!("");
info!("// Send local changes");
info!("loop {{");
info!(" let change = get_local_change().await;");
info!(" tx.send(&change).await?;");
info!("}}");
Ok(())
}
pub async fn error_handling_example() -> Result<()> {
info!("Example 6: Error handling in streaming RPC");
info!("\nServer side - Error handling:");
info!("ctx.stream_respond(\"/data/process\", |req, sender| async move {{");
info!(" for item in req.items {{");
info!(" match process_item(&item).await {{");
info!(" Ok(result) => {{");
info!(" sender.send(&result).await?;");
info!(" }}");
info!(" Err(e) => {{");
info!(" // Send error and terminate stream");
info!(" sender.error(&format!(\"Processing failed: {{}}\", e)).await?;");
info!(" return Ok(());");
info!(" }}");
info!(" }}");
info!(" }}");
info!(" sender.end().await?;");
info!(" Ok(())");
info!("}}).await?;");
info!("\nClient side - Detect errors:");
info!("let mut stream = ctx.stream_request(\"/data/process\", &req).await?;");
info!("");
info!("while let Some(result) = stream.recv().await {{");
info!(" process_result(&result);");
info!("}}");
info!("");
info!("if stream.is_ended() {{");
info!(" println!(\"Stream ended (could be normal or error)\");");
info!("}}");
Ok(())
}
pub async fn lifecycle_management_example() -> Result<()> {
info!("Example 7: Stream lifecycle management");
info!("\nStream lifecycle:");
info!("1. Client subscribes to response topic");
info!("2. Client sends initial request with stream_id");
info!("3. Server receives request, creates sender for that stream_id");
info!("4. Server sends multiple StreamMessage<T> with:");
info!(" - stream_id (for filtering)");
info!(" - sequence number (for ordering)");
info!(" - payload (Data, End, Error, Heartbeat)");
info!("5. Client receives messages, filters by stream_id");
info!("6. Stream ends with End or Error message");
info!("");
info!("Stream message structure:");
info!("StreamMessage {{");
info!(" stream_id: \"uuid-v4\",");
info!(" sequence: 0..N,");
info!(" payload: StreamPayload::Data(T) | End | Error | Heartbeat,");
info!(" timestamp: unix_micros,");
info!("}}");
Ok(())
}
pub async fn concurrent_streams_example() -> Result<()> {
info!("Example 8: Multiple concurrent streams");
info!("\nMultiple streams example:");
info!("// Start multiple streams concurrently");
info!("let camera1 = ctx.stream_request(\"/camera/1/feed\", &()).await?;");
info!("let camera2 = ctx.stream_request(\"/camera/2/feed\", &()).await?;");
info!("let lidar = ctx.stream_request(\"/lidar/scan\", &()).await?;");
info!("");
info!("// Each stream has unique stream_id for filtering");
info!("// Messages are delivered via pub/sub but filtered by stream_id");
info!("");
info!("tokio::select! {{");
info!(" Some(frame) = camera1.recv() => process_camera1(frame),");
info!(" Some(frame) = camera2.recv() => process_camera2(frame),");
info!(" Some(scan) = lidar.recv() => process_lidar(scan),");
info!("}}");
Ok(())
}
fn main() -> Result<()> {
tracing_subscriber::fmt().with_max_level(tracing::Level::INFO).init();
println!("Streaming RPC Usage Examples");
println!("=============================\n");
println!("Streaming RPC enables long-running operations with multiple messages:");
println!("- Server streaming: One request, multiple responses");
println!("- Client streaming: Multiple requests, one response");
println!("- Bidirectional: Both sides send multiple messages");
println!();
let runtime = tokio::runtime::Runtime::new().unwrap();
runtime.block_on(async {
server_streaming_video_example().await?;
println!();
server_streaming_telemetry_example().await?;
println!();
server_streaming_progress_example().await?;
println!();
bidirectional_chat_example().await?;
println!();
bidirectional_collaboration_example().await?;
println!();
error_handling_example().await?;
println!();
lifecycle_management_example().await?;
println!();
concurrent_streams_example().await?;
println!();
Ok::<(), Mecha10Error>(())
})?;
println!("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
println!("All examples completed!");
println!("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n");
println!("Key concepts:");
println!(" • Stream IDs: Unique UUID per stream for filtering");
println!(" • Sequence numbers: Ordering and duplicate detection");
println!(" • Stream lifecycle: Start → Data* → (End | Error)");
println!(" • Heartbeats: Keep-alive for long-running streams");
println!(" • Error propagation: Errors terminate stream gracefully");
println!();
println!("Use cases:");
println!(" ✓ Video/audio streaming");
println!(" ✓ Sensor telemetry");
println!(" ✓ Progress updates for long tasks");
println!(" ✓ Interactive chat/collaboration");
println!(" ✓ File transfer (chunked)");
println!(" ✓ Real-time monitoring");
Ok(())
}