1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
//! Receiver for subscribing to messages
//!
//! Provides a type-safe wrapper around tokio channels for receiving messages
//! from subscribed topics with support for both bounded and unbounded channels.
//!
//! # Backpressure Strategies
//!
//! The receiver supports multiple backpressure strategies to prevent memory exhaustion
//! from fast publishers. Choose based on your use case:
//!
//! ## 1. Unbounded (Default)
//! No backpressure - messages queue indefinitely. Use for low-rate topics.
//!
//! ```rust,no_run
//! # use mecha10_core::prelude::*;
//! # use mecha10_core::topics::sensor;
//! # async fn example(ctx: &Context) -> Result<()> {
//! let mut rx = ctx.subscribe(sensor::CAMERA_RGB).await?;
//! // Unlimited queue - risk of memory growth
//! # Ok(())
//! # }
//! ```
//!
//! ## 2. Bounded with Blocking (`.with_capacity()`)
//! Publisher waits when buffer full. Use when all messages are important.
//!
//! ```rust,no_run
//! # use mecha10_core::prelude::*;
//! # use mecha10_core::topics::sensor;
//! # async fn example(ctx: &Context) -> Result<()> {
//! let mut rx = ctx.subscribe(sensor::CAMERA_RGB)
//! .await?
//! .with_capacity(50); // Block publisher when 50 messages queued
//! # Ok(())
//! # }
//! ```
//!
//! ## 3. Bounded with Drop Oldest (`.with_drop_oldest()`)
//! Drops old messages when buffer full. Use for real-time sensor data.
//!
//! ```rust,no_run
//! # use mecha10_core::prelude::*;
//! # use mecha10_core::topics::sensor;
//! # async fn example(ctx: &Context) -> Result<()> {
//! let mut rx = ctx.subscribe(sensor::LIDAR_SCAN)
//! .await?
//! .with_drop_oldest(10); // Keep only 10 newest scans
//! # Ok(())
//! # }
//! ```
//!
//! ## 4. Rate Limiting (`.debounce()`)
//! Limit emission rate. Use to reduce update frequency.
//!
//! ```rust,no_run
//! # use mecha10_core::prelude::*;
//! # use mecha10_core::topics::sensor;
//! # use std::time::Duration;
//! # async fn example(ctx: &Context) -> Result<()> {
//! let mut rx = ctx.subscribe(sensor::CAMERA_RGB)
//! .await?
//! .debounce(Duration::from_millis(100)); // Max 10 Hz
//! # Ok(())
//! # }
//! ```
//!
//! ## Stream Combinators
//!
//! Transform and combine streams with powerful combinators:
//!
//! ```rust,no_run
//! # use mecha10_core::prelude::*;
//! # use mecha10_core::messages::{Image, LaserScan, Imu};
//! # use mecha10_core::topics::sensor;
//! # use std::time::Duration;
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Transform messages
//! let mut scaled = ctx.subscribe(sensor::CAMERA_RGB)
//! .await?
//! .map(|img| scale_down(&img));
//!
//! // Filter messages
//! let mut valid = ctx.subscribe(sensor::LIDAR_SCAN)
//! .await?
//! .filter(|scan| scan.range_min > 0.1);
//!
//! // Combine streams (sensor fusion)
//! let camera = ctx.subscribe(sensor::CAMERA_RGB).await?;
//! let lidar = ctx.subscribe(sensor::LIDAR_SCAN).await?;
//! let mut fused = camera.zip(lidar);
//!
//! // Window messages over time (for aggregation)
//! let mut windowed = ctx.subscribe(sensor::IMU)
//! .await?
//! .window(Duration::from_millis(100)) // Collect 100ms batches
//! .map(|batch| calculate_average(&batch)); // Aggregate
//!
//! // Chain combinators
//! let mut processed = ctx.subscribe(sensor::CAMERA_RGB)
//! .await?
//! .with_drop_oldest(20) // Buffer 20 latest
//! .debounce(Duration::from_millis(100)) // Max 10 Hz
//! .filter(|img| img.width == 640) // Only 640px wide
//! .map(|img| preprocess(&img)); // Transform
//! # Ok(())
//! # }
//! # fn scale_down(img: &Image) -> Image { img.clone() }
//! # fn calculate_average(batch: &[Imu]) -> Imu { batch[0].clone() }
//! # fn preprocess(img: &Image) -> Image { img.clone() }
//! ```
//!
//! # Monitoring Backpressure
//!
//! Use `.len()`, `.capacity()`, and `.utilization()` to monitor queue health:
//!
//! ```rust,no_run
//! # use mecha10_core::prelude::*;
//! # async fn example(rx: &mut Receiver<String>) {
//! if let Some(util) = rx.utilization() {
//! if util > 0.8 {
//! println!("WARNING: Queue is {}% full!", util * 100.0);
//! }
//! }
//! # }
//! ```
//!
//! This module is organized into:
//! - `drop_oldest`: The bounded, oldest-evicting buffer backing `with_drop_oldest()`
//! - `types`: The `Receiver`/`ReceiverInner` types and their basic operations
//! - `relay`: Shared scaffolding for combinators that relay messages into a new channel
//! - `combinators`: `map`, `filter`, `filter_map`
//! - `time_combinators`: `debounce`, `window`
//! - `zip`: Combining two receivers into one paired stream
pub use Receiver;
pub use ReceiverInner;