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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
use std::{task::Poll, time::Duration};
/// Subscriber-side preferences for receiving a track.
///
/// Each subscriber holds its own [`Subscription`]; the publisher observes an
/// aggregate across all live subscribers via [`crate::track::Producer::subscription`].
/// A subscriber can change its preferences after the fact with
/// [`crate::track::Subscriber::update`].
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct Subscription {
/// Delivery priority. Higher values preempt lower ones when bandwidth is constrained.
pub priority: u8,
/// Whether groups are prioritized in sequence order. Groups may always arrive
/// out-of-order (or not at all) over the network. Defaults to `false`; the
/// aggregate is ordered only when every live subscriber asks for it.
pub ordered: bool,
/// The maximum age of a non-latest group before it is skipped. `Duration::ZERO`
/// skips immediately (e.g. group 8 arriving means group 7 is skipped); a larger
/// value tolerates that much reordering before giving up on the older group.
///
/// This is the `Subscriber Max Latency` on the wire, enforced by the publisher's
/// cache. Receivers that buffer (e.g. a jitter buffer) enforce the same budget
/// locally, and a group is skipped once either measure exceeds it.
pub latency_max: Duration,
/// First group the publisher should deliver, or `None` to start at the latest group.
///
/// A request, aggregated across every live subscriber (the earliest explicit start
/// wins), so it says what the publisher sends, not what any one subscriber sees.
/// [`crate::track::Subscriber::start_at`] is the local read cursor; setting one does
/// not imply the other. See [Local cursor vs wire
/// preference](crate::track::Subscriber#local-cursor-vs-wire-preference).
pub group_start: Option<u64>,
/// Last group the publisher should deliver (inclusive), or `None` for no end.
///
/// A request, aggregated across every live subscriber (any unbounded subscriber makes
/// the aggregate unbounded). [`crate::track::Subscriber::end_at`] is the local read
/// cursor; setting one does not imply the other. See [Local cursor vs wire
/// preference](crate::track::Subscriber#local-cursor-vs-wire-preference).
pub group_end: Option<u64>,
}
impl Default for Subscription {
fn default() -> Self {
Self {
priority: 0,
ordered: false,
latency_max: Duration::ZERO,
group_start: None,
group_end: None,
}
}
}
impl Subscription {
/// Set the delivery priority, returning `self` for chaining.
pub fn with_priority(mut self, priority: u8) -> Self {
self.priority = priority;
self
}
/// Set whether groups are prioritized in sequence order, returning `self` for
/// chaining. Groups may always arrive out-of-order (or not at all) over the network.
pub fn with_ordered(mut self, ordered: bool) -> Self {
self.ordered = ordered;
self
}
/// Set the maximum age of a non-latest group before it is skipped, returning
/// `self` for chaining.
pub fn with_latency_max(mut self, latency_max: Duration) -> Self {
self.latency_max = latency_max;
self
}
/// Set the first group to deliver, returning `self` for chaining.
pub fn with_group_start(mut self, group_start: impl Into<Option<u64>>) -> Self {
self.group_start = group_start.into();
self
}
/// Set the last group to deliver (inclusive), returning `self` for chaining.
pub fn with_group_end(mut self, group_end: impl Into<Option<u64>>) -> Self {
self.group_end = group_end.into();
self
}
// Fold this subscription into the running aggregate: Ready with the merged
// result when it demands more than `combined`, Pending when it's a subset
// (so callers can skip a redundant broadcast of the same aggregate).
pub(super) fn poll_combined(&self, combined: &Option<Subscription>) -> Poll<Subscription> {
let Some(combined) = combined else {
return Poll::Ready(self.clone());
};
let merged = Subscription {
priority: self.priority.max(combined.priority),
// Sequence-first prioritization is enabled only when every subscriber wants it.
ordered: self.ordered && combined.ordered,
latency_max: self.latency_max.max(combined.latency_max),
group_start: min_some(self.group_start, combined.group_start),
group_end: max_unbounded(self.group_end, combined.group_end),
};
if &merged != combined {
return Poll::Ready(merged);
}
Poll::Pending
}
}
/// The lower of two optional bounds, treating `None` as unbounded.
pub(super) fn min_some(a: Option<u64>, b: Option<u64>) -> Option<u64> {
match (a, b) {
(Some(a), Some(b)) => Some(a.min(b)),
(Some(a), None) | (None, Some(a)) => Some(a),
(None, None) => None,
}
}
fn max_unbounded(a: Option<u64>, b: Option<u64>) -> Option<u64> {
match (a, b) {
(Some(a), Some(b)) => Some(a.max(b)),
(None, _) | (_, None) => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn combine(subscriptions: &[Subscription]) -> Option<Subscription> {
let mut combined = None;
for sub in subscriptions {
if let Poll::Ready(merged) = sub.poll_combined(&combined) {
combined = Some(merged);
}
}
combined
}
#[test]
fn combined_ordered_stays_ordered_for_multiple_ordered_viewers() {
let subscription = Subscription::default().with_ordered(true);
let combined = combine(&[subscription.clone(), subscription.clone(), subscription]).unwrap();
assert!(combined.ordered);
}
#[test]
fn combined_any_unordered_viewer_disables_ordered() {
let unordered = Subscription::default().with_ordered(false);
let ordered = Subscription::default().with_ordered(true);
let combined = combine(&[unordered, ordered]).unwrap();
assert!(!combined.ordered);
}
#[test]
fn combined_group_start_uses_earliest_explicit_start() {
let live = Subscription::default().with_group_start(None);
let catchup = Subscription::default().with_group_start(10);
let older_catchup = Subscription::default().with_group_start(5);
let combined = combine(&[live, catchup, older_catchup]).unwrap();
assert_eq!(combined.group_start, Some(5));
}
#[test]
fn combined_group_end_keeps_live_subscription_unbounded() {
let live = Subscription::default().with_group_end(None);
let bounded = Subscription::default().with_group_end(10);
let combined = combine(&[live, bounded]).unwrap();
assert_eq!(combined.group_end, None);
}
#[test]
fn combined_group_end_uses_latest_bounded_end() {
let early = Subscription::default().with_group_end(10);
let late = Subscription::default().with_group_end(20);
let combined = combine(&[early, late]).unwrap();
assert_eq!(combined.group_end, Some(20));
}
}