|
1 |
+ |
//! In-process fan-out.
|
|
2 |
+ |
//!
|
|
3 |
+ |
//! One [`tokio::sync::broadcast`] channel per room, created when the first
|
|
4 |
+ |
//! listener arrives and pruned when the last one leaves. This is correct for the
|
|
5 |
+ |
//! deployment both consumers actually have (a single process on a single box) and
|
|
6 |
+ |
//! is knowingly the wrong shape for more than one node. Going multi-node means
|
|
7 |
+ |
//! putting Postgres `LISTEN/NOTIFY` or a message bus behind [`Hub::publish`]; it
|
|
8 |
+ |
//! is a second entry on the same list as the in-process rate limiter, not a
|
|
9 |
+ |
//! surprise.
|
|
10 |
+ |
//!
|
|
11 |
+ |
//! # Why the connection caps are load-bearing
|
|
12 |
+ |
//!
|
|
13 |
+ |
//! Both units run under `MemoryMax=512M`, a hard cgroup cap. Each listener holds
|
|
14 |
+ |
//! a task, a buffer, and a broadcast receiver, and an OOM does not degrade chat,
|
|
15 |
+ |
//! it restarts the whole site. So the caps here are a availability control, not
|
|
16 |
+ |
//! tidiness, and they are enforced before any allocation the connection would
|
|
17 |
+ |
//! own.
|
|
18 |
+ |
//!
|
|
19 |
+ |
//! # Divergence from the SyncKit SSE pattern
|
|
20 |
+ |
//!
|
|
21 |
+ |
//! This is modeled on `server/src/routes/synckit/subscribe.rs` and deviates from
|
|
22 |
+ |
//! it twice, both deliberately:
|
|
23 |
+ |
//!
|
|
24 |
+ |
//! 1. **Pruning is atomic.** SyncKit reads `receiver_count()`, drops the guard,
|
|
25 |
+ |
//! then calls `remove()`. Between those two steps a new subscriber can take
|
|
26 |
+ |
//! the entry, and the `remove()` then drops a channel that has a live
|
|
27 |
+ |
//! listener; the next publish lazily creates a *different* channel and the
|
|
28 |
+ |
//! listener silently receives nothing. SyncKit tolerates this because its
|
|
29 |
+ |
//! clients re-pull on their own schedule. A chat client would just sit in a
|
|
30 |
+ |
//! dead room. [`DashMap::remove_if`] runs the predicate under the shard lock,
|
|
31 |
+ |
//! which closes the race and still avoids the same-shard deadlock that made
|
|
32 |
+ |
//! SyncKit split the read and the remove in the first place.
|
|
33 |
+ |
//! 2. **Lag is surfaced, not swallowed.** SyncKit discards `Lagged`. Here it
|
|
34 |
+ |
//! becomes [`ChatEvent::Gap`] so the client re-fetches from its cursor,
|
|
35 |
+ |
//! because a room with a silent hole in it is worse than a visible one.
|
|
36 |
+ |
|
|
37 |
+ |
use std::sync::Arc;
|
|
38 |
+ |
use std::sync::atomic::{AtomicUsize, Ordering};
|
|
39 |
+ |
|
|
40 |
+ |
use dashmap::DashMap;
|
|
41 |
+ |
use tokio::sync::broadcast;
|
|
42 |
+ |
|
|
43 |
+ |
use crate::error::{ChatError, LimitScope};
|
|
44 |
+ |
use crate::event::ChatEvent;
|
|
45 |
+ |
use crate::ids::{RoomId, UserId};
|
|
46 |
+ |
|
|
47 |
+ |
/// Caps and buffer sizes. Every consumer sets these explicitly; there is no
|
|
48 |
+ |
/// unbounded configuration.
|
|
49 |
+ |
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
50 |
+ |
pub struct HubLimits {
|
|
51 |
+ |
/// Concurrent listeners one user may hold. A few, for multiple tabs. Not a
|
|
52 |
+ |
/// spam control, a fairness one: without it a single client can consume the
|
|
53 |
+ |
/// global budget.
|
|
54 |
+ |
pub max_connections_per_user: usize,
|
|
55 |
+ |
/// Concurrent listeners across every room in the process. This is the number
|
|
56 |
+ |
/// that keeps the box alive.
|
|
57 |
+ |
pub max_connections_total: usize,
|
|
58 |
+ |
/// Messages buffered per room before a slow listener starts missing events
|
|
59 |
+ |
/// and gets a [`ChatEvent::Gap`].
|
|
60 |
+ |
pub room_buffer: usize,
|
|
61 |
+ |
}
|
|
62 |
+ |
|
|
63 |
+ |
impl Default for HubLimits {
|
|
64 |
+ |
fn default() -> Self {
|
|
65 |
+ |
Self {
|
|
66 |
+ |
max_connections_per_user: 4,
|
|
67 |
+ |
max_connections_total: 2_000,
|
|
68 |
+ |
room_buffer: 256,
|
|
69 |
+ |
}
|
|
70 |
+ |
}
|
|
71 |
+ |
}
|
|
72 |
+ |
|
|
73 |
+ |
struct Inner {
|
|
74 |
+ |
rooms: DashMap<RoomId, broadcast::Sender<ChatEvent>>,
|
|
75 |
+ |
per_user: DashMap<UserId, AtomicUsize>,
|
|
76 |
+ |
total: AtomicUsize,
|
|
77 |
+ |
limits: HubLimits,
|
|
78 |
+ |
}
|
|
79 |
+ |
|
|
80 |
+ |
/// Per-room fan-out with connection accounting.
|
|
81 |
+ |
///
|
|
82 |
+ |
/// Cheap to clone; both consumers hang one off application state.
|
|
83 |
+ |
#[derive(Clone)]
|
|
84 |
+ |
pub struct Hub {
|
|
85 |
+ |
inner: Arc<Inner>,
|
|
86 |
+ |
}
|
|
87 |
+ |
|
|
88 |
+ |
impl Hub {
|
|
89 |
+ |
pub fn new(limits: HubLimits) -> Self {
|
|
90 |
+ |
Self {
|
|
91 |
+ |
inner: Arc::new(Inner {
|
|
92 |
+ |
rooms: DashMap::new(),
|
|
93 |
+ |
per_user: DashMap::new(),
|
|
94 |
+ |
total: AtomicUsize::new(0),
|
|
95 |
+ |
limits,
|
|
96 |
+ |
}),
|
|
97 |
+ |
}
|
|
98 |
+ |
}
|
|
99 |
+ |
|
|
100 |
+ |
/// Send to everyone listening to `room`, returning how many received it.
|
|
101 |
+ |
///
|
|
102 |
+ |
/// Zero is normal and not an error: an empty room is the common case, and a
|
|
103 |
+ |
/// message still lands in storage regardless of who is connected.
|
|
104 |
+ |
pub fn publish(&self, room: RoomId, event: ChatEvent) -> usize {
|
|
105 |
+ |
self.inner
|
|
106 |
+ |
.rooms
|
|
107 |
+ |
.get(&room)
|
|
108 |
+ |
.and_then(|tx| tx.send(event).ok())
|
|
109 |
+ |
.unwrap_or(0)
|
|
110 |
+ |
}
|
|
111 |
+ |
|
|
112 |
+ |
/// Start listening to `room`.
|
|
113 |
+ |
///
|
|
114 |
+ |
/// # Errors
|
|
115 |
+ |
///
|
|
116 |
+ |
/// [`ChatError::ConnectionLimit`] when the user or the process is at capacity.
|
|
117 |
+ |
/// The caller has already passed authz by this point, so this is a resource
|
|
118 |
+ |
/// refusal and should read as one: it is a 503, not a 403.
|
|
119 |
+ |
pub fn subscribe(&self, room: RoomId, user: UserId) -> Result<Subscription, ChatError> {
|
|
120 |
+ |
// The permit is acquired first and owns the rollback. There is no window
|
|
121 |
+ |
// in which a counter is incremented but nothing is responsible for
|
|
122 |
+ |
// decrementing it, including if anything below panics.
|
|
123 |
+ |
let permit = ConnectionPermit::acquire(self.inner.clone(), user)?;
|
|
124 |
+ |
|
|
125 |
+ |
let rx = self
|
|
126 |
+ |
.inner
|
|
127 |
+ |
.rooms
|
|
128 |
+ |
.entry(room)
|
|
129 |
+ |
.or_insert_with(|| broadcast::channel(self.inner.limits.room_buffer).0)
|
|
130 |
+ |
.subscribe();
|
|
131 |
+ |
|
|
132 |
+ |
Ok(Subscription {
|
|
133 |
+ |
rx,
|
|
134 |
+ |
room,
|
|
135 |
+ |
_permit: permit,
|
|
136 |
+ |
})
|
|
137 |
+ |
}
|
|
138 |
+ |
|
|
139 |
+ |
/// Rooms currently holding a channel. Diagnostics only.
|
|
140 |
+ |
pub fn room_count(&self) -> usize {
|
|
141 |
+ |
self.inner.rooms.len()
|
|
142 |
+ |
}
|
|
143 |
+ |
|
|
144 |
+ |
/// Listeners across every room. Worth exposing in a health payload: it is the
|
|
145 |
+ |
/// number that predicts the memory cap.
|
|
146 |
+ |
pub fn connection_count(&self) -> usize {
|
|
147 |
+ |
self.inner.total.load(Ordering::Acquire)
|
|
148 |
+ |
}
|
|
149 |
+ |
|
|
150 |
+ |
/// Listeners in one room.
|
|
151 |
+ |
pub fn room_connection_count(&self, room: RoomId) -> usize {
|
|
152 |
+ |
self.inner
|
|
153 |
+ |
.rooms
|
|
154 |
+ |
.get(&room)
|
|
155 |
+ |
.map(|tx| tx.receiver_count())
|
|
156 |
+ |
.unwrap_or(0)
|
|
157 |
+ |
}
|
|
158 |
+ |
}
|
|
159 |
+ |
|
|
160 |
+ |
/// Holds one connection's slot in the global and per-user counters, releasing
|
|
161 |
+ |
/// both on drop.
|
|
162 |
+ |
struct ConnectionPermit {
|
|
163 |
+ |
inner: Arc<Inner>,
|
|
164 |
+ |
user: UserId,
|
|
165 |
+ |
}
|
|
166 |
+ |
|
|
167 |
+ |
impl ConnectionPermit {
|
|
168 |
+ |
fn acquire(inner: Arc<Inner>, user: UserId) -> Result<Self, ChatError> {
|
|
169 |
+ |
// Global first: it is a single atomic, so the cheap check rejects the
|
|
170 |
+ |
// overload case without touching the per-user map at all.
|
|
171 |
+ |
let prev = inner.total.fetch_add(1, Ordering::AcqRel);
|
|
172 |
+ |
if prev >= inner.limits.max_connections_total {
|
|
173 |
+ |
inner.total.fetch_sub(1, Ordering::AcqRel);
|
|
174 |
+ |
return Err(ChatError::ConnectionLimit(LimitScope::Global));
|
|
175 |
+ |
}
|
|
176 |
+ |
|
|
177 |
+ |
let over_user_limit = {
|
|
178 |
+ |
let entry = inner.per_user.entry(user).or_insert_with(|| AtomicUsize::new(0));
|
|
179 |
+ |
let prev = entry.value().fetch_add(1, Ordering::AcqRel);
|
|
180 |
+ |
if prev >= inner.limits.max_connections_per_user {
|
|
181 |
+ |
entry.value().fetch_sub(1, Ordering::AcqRel);
|
|
182 |
+ |
true
|
|
183 |
+ |
} else {
|
|
184 |
+ |
false
|
|
185 |
+ |
}
|
|
186 |
+ |
};
|
|
187 |
+ |
|
|
188 |
+ |
if over_user_limit {
|
|
189 |
+ |
inner.total.fetch_sub(1, Ordering::AcqRel);
|
|
190 |
+ |
// The entry may now be at zero and prunable, but leaving it costs one
|
|
191 |
+ |
// small map slot and the next successful subscribe reuses it. Do not
|
|
192 |
+ |
// remove here: this path is reached under contention, which is
|
|
193 |
+ |
// exactly when an extra shard lock is least welcome.
|
|
194 |
+ |
return Err(ChatError::ConnectionLimit(LimitScope::User));
|
|
195 |
+ |
}
|
|
196 |
+ |
|
|
197 |
+ |
Ok(Self { inner, user })
|
|
198 |
+ |
}
|
|
199 |
+ |
}
|
|
200 |
+ |
|
|
201 |
+ |
impl Drop for ConnectionPermit {
|
|
202 |
+ |
fn drop(&mut self) {
|
|
203 |
+ |
self.inner.total.fetch_sub(1, Ordering::AcqRel);
|
|
204 |
+ |
|
|
205 |
+ |
if let Some(entry) = self.inner.per_user.get(&self.user) {
|
|
206 |
+ |
entry.value().fetch_sub(1, Ordering::AcqRel);
|
|
207 |
+ |
}
|
|
208 |
+ |
// Predicate runs under the shard lock, so a user reconnecting in this
|
|
209 |
+ |
// instant either keeps their entry or gets a fresh one. Never both.
|
|
210 |
+ |
self.inner
|
|
211 |
+ |
.per_user
|
|
212 |
+ |
.remove_if(&self.user, |_, count| count.load(Ordering::Acquire) == 0);
|
|
213 |
+ |
}
|
|
214 |
+ |
}
|
|
215 |
+ |
|
|
216 |
+ |
/// A live listener on one room.
|
|
217 |
+ |
///
|
|
218 |
+ |
/// Dropping it releases the connection slot and prunes the room channel if this
|
|
219 |
+ |
/// was the last listener.
|
|
220 |
+ |
pub struct Subscription {
|
|
221 |
+ |
rx: broadcast::Receiver<ChatEvent>,
|
|
222 |
+ |
room: RoomId,
|
|
223 |
+ |
_permit: ConnectionPermit,
|
|
224 |
+ |
}
|
|
225 |
+ |
|
|
226 |
+ |
impl Subscription {
|
|
227 |
+ |
/// Next event, or `None` once the room is gone and nothing more will arrive.
|
|
228 |
+ |
///
|
|
229 |
+ |
/// A listener that fell behind gets [`ChatEvent::Gap`] rather than silence.
|
|
230 |
+ |
/// The events it missed are not recoverable from the channel by design: the
|
|
231 |
+ |
/// client re-fetches from its cursor, which is the same path a reconnect
|
|
232 |
+ |
/// after a deploy takes.
|
|
233 |
+ |
pub async fn recv(&mut self) -> Option<ChatEvent> {
|
|
234 |
+ |
match self.rx.recv().await {
|
|
235 |
+ |
Ok(event) => Some(event),
|
|
236 |
+ |
Err(broadcast::error::RecvError::Lagged(_)) => Some(ChatEvent::Gap),
|
|
237 |
+ |
Err(broadcast::error::RecvError::Closed) => None,
|
|
238 |
+ |
}
|
|
239 |
+ |
}
|
|
240 |
+ |
|
|
241 |
+ |
pub fn room(&self) -> RoomId {
|
|
242 |
+ |
self.room
|
|
243 |
+ |
}
|
|
244 |
+ |
}
|
|
245 |
+ |
|
|
246 |
+ |
impl Drop for Subscription {
|
|
247 |
+ |
fn drop(&mut self) {
|
|
248 |
+ |
// `self.rx` is still alive here (fields drop after this body), so this
|
|
249 |
+ |
// subscription is still counted. Compare against 1, not 0.
|
|
250 |
+ |
//
|
|
251 |
+ |
// The predicate runs under the shard lock, so a subscriber arriving in
|
|
252 |
+ |
// this instant either keeps this channel or creates a fresh one after the
|
|
253 |
+ |
// removal. It cannot end up holding a receiver on a sender that has just
|
|
254 |
+ |
// been dropped from the map, which is the silent-dead-room failure.
|
|
255 |
+ |
self._permit
|
|
256 |
+ |
.inner
|
|
257 |
+ |
.rooms
|
|
258 |
+ |
.remove_if(&self.room, |_, tx| tx.receiver_count() <= 1);
|
|
259 |
+ |
}
|
|
260 |
+ |
}
|
|
261 |
+ |
|
|
262 |
+ |
#[cfg(test)]
|
|
263 |
+ |
mod tests {
|
|
264 |
+ |
use super::*;
|
|
265 |
+ |
use crate::ids::MessageId;
|
|
266 |
+ |
use uuid::Uuid;
|
|
267 |
+ |
|
|
268 |
+ |
fn hub(limits: HubLimits) -> Hub {
|
|
269 |
+ |
Hub::new(limits)
|
|
270 |
+ |
}
|
|
271 |
+ |
|
|
272 |
+ |
fn room() -> RoomId {
|
|
273 |
+ |
RoomId(Uuid::new_v4())
|
|
274 |
+ |
}
|
|
275 |
+ |
|
|
276 |
+ |
fn user() -> UserId {
|
|
277 |
+ |
UserId(Uuid::new_v4())
|
|
278 |
+ |
}
|
|
279 |
+ |
|
|
280 |
+ |
fn delete(id: i64) -> ChatEvent {
|
|
281 |
+ |
ChatEvent::Delete {
|
|
282 |
+ |
id: MessageId(id),
|
|
283 |
+ |
}
|
|
284 |
+ |
}
|
|
285 |
+ |
|
|
286 |
+ |
#[tokio::test]
|
|
287 |
+ |
async fn delivers_to_every_listener_in_the_room() {
|
|
288 |
+ |
let hub = hub(HubLimits::default());
|
|
289 |
+ |
let r = room();
|
|
290 |
+ |
let mut a = hub.subscribe(r, user()).unwrap();
|
|
291 |
+ |
let mut b = hub.subscribe(r, user()).unwrap();
|
|
292 |
+ |
|
|
293 |
+ |
assert_eq!(hub.publish(r, delete(1)), 2);
|
|
294 |
+ |
assert_eq!(a.recv().await, Some(delete(1)));
|
|
295 |
+ |
assert_eq!(b.recv().await, Some(delete(1)));
|
|
296 |
+ |
}
|
|
297 |
+ |
|
|
298 |
+ |
#[tokio::test]
|
|
299 |
+ |
async fn rooms_are_isolated() {
|
|
300 |
+ |
let hub = hub(HubLimits::default());
|
|
301 |
+ |
let (r1, r2) = (room(), room());
|
|
302 |
+ |
let mut listener = hub.subscribe(r1, user()).unwrap();
|
|
303 |
+ |
|
|
304 |
+ |
assert_eq!(hub.publish(r2, delete(1)), 0);
|
|
305 |
+ |
assert_eq!(hub.publish(r1, delete(2)), 1);
|
|
306 |
+ |
assert_eq!(listener.recv().await, Some(delete(2)));
|
|
307 |
+ |
}
|
|
308 |
+ |
|
|
309 |
+ |
#[test]
|
|
310 |
+ |
fn publishing_to_an_empty_room_is_not_an_error() {
|
|
311 |
+ |
let hub = hub(HubLimits::default());
|
|
312 |
+ |
assert_eq!(hub.publish(room(), delete(1)), 0);
|
|
313 |
+ |
// and does not conjure a channel for a room nobody is listening to
|
|
314 |
+ |
assert_eq!(hub.room_count(), 0);
|
|
315 |
+ |
}
|
|
316 |
+ |
|
|
317 |
+ |
#[test]
|
|
318 |
+ |
fn room_channel_is_pruned_when_the_last_listener_leaves() {
|
|
319 |
+ |
let hub = hub(HubLimits::default());
|
|
320 |
+ |
let r = room();
|
|
321 |
+ |
|
|
322 |
+ |
let a = hub.subscribe(r, user()).unwrap();
|
|
323 |
+ |
let b = hub.subscribe(r, user()).unwrap();
|
|
324 |
+ |
assert_eq!(hub.room_count(), 1);
|
|
325 |
+ |
|
|
326 |
+ |
drop(a);
|
|
327 |
+ |
assert_eq!(hub.room_count(), 1, "one listener remains");
|
|
328 |
+ |
|
|
329 |
+ |
drop(b);
|
|
330 |
+ |
assert_eq!(hub.room_count(), 0, "last listener pruned the room");
|
|
331 |
+ |
}
|
|
332 |
+ |
|
|
333 |
+ |
#[tokio::test]
|
|
334 |
+ |
async fn a_room_recreated_after_pruning_still_delivers() {
|
|
335 |
+ |
// The failure this guards: pruning a channel that a new subscriber has
|
|
336 |
+ |
// just taken, leaving them attached to an orphaned sender.
|
|
337 |
+ |
let hub = hub(HubLimits::default());
|
|
338 |
+ |
let r = room();
|
|
339 |
+ |
|
|
340 |
+ |
let first = hub.subscribe(r, user()).unwrap();
|
|
341 |
+ |
drop(first);
|
|
342 |
+ |
assert_eq!(hub.room_count(), 0);
|
|
343 |
+ |
|
|
344 |
+ |
let mut second = hub.subscribe(r, user()).unwrap();
|
|
345 |
+ |
assert_eq!(hub.publish(r, delete(9)), 1);
|
|
346 |
+ |
assert_eq!(second.recv().await, Some(delete(9)));
|
|
347 |
+ |
}
|
|
348 |
+ |
|
|
349 |
+ |
#[test]
|
|
350 |
+ |
fn per_user_limit_is_enforced_and_released() {
|
|
351 |
+ |
let hub = hub(HubLimits {
|
|
352 |
+ |
max_connections_per_user: 2,
|
|
353 |
+ |
..HubLimits::default()
|
|
354 |
+ |
});
|
|
355 |
+ |
let r = room();
|
|
356 |
+ |
let u = user();
|
|
357 |
+ |
|
|
358 |
+ |
let a = hub.subscribe(r, u).unwrap();
|
|
359 |
+ |
let b = hub.subscribe(r, u).unwrap();
|
|
360 |
+ |
assert!(matches!(
|
|
361 |
+ |
hub.subscribe(r, u),
|
|
362 |
+ |
Err(ChatError::ConnectionLimit(LimitScope::User))
|
|
363 |
+ |
));
|
|
364 |
+ |
|
|
365 |
+ |
// A different user is unaffected by another's exhausted budget.
|
|
366 |
+ |
let other = hub.subscribe(r, user()).unwrap();
|
|
367 |
+ |
|
|
368 |
+ |
drop(a);
|
|
369 |
+ |
let _c = hub.subscribe(r, u).expect("slot released on drop");
|
|
370 |
+ |
drop((b, other));
|
|
371 |
+ |
}
|
|
372 |
+ |
|
|
373 |
+ |
#[test]
|
|
374 |
+ |
fn global_limit_is_enforced_and_released() {
|
|
375 |
+ |
let hub = hub(HubLimits {
|
|
376 |
+ |
max_connections_total: 2,
|
|
377 |
+ |
..HubLimits::default()
|
|
378 |
+ |
});
|
|
379 |
+ |
let r = room();
|
|
380 |
+ |
|
|
381 |
+ |
let a = hub.subscribe(r, user()).unwrap();
|
|
382 |
+ |
let b = hub.subscribe(r, user()).unwrap();
|
|
383 |
+ |
assert!(matches!(
|
|
384 |
+ |
hub.subscribe(r, user()),
|
|
385 |
+ |
Err(ChatError::ConnectionLimit(LimitScope::Global))
|
|
386 |
+ |
));
|
|
387 |
+ |
|
|
388 |
+ |
drop(a);
|
|
389 |
+ |
assert_eq!(hub.connection_count(), 1);
|
|
390 |
+ |
let _c = hub.subscribe(r, user()).expect("slot released on drop");
|
|
391 |
+ |
drop(b);
|
|
392 |
+ |
}
|
|
393 |
+ |
|
|
394 |
+ |
#[test]
|
|
395 |
+ |
fn a_rejected_subscribe_leaks_no_slot() {
|
|
396 |
+ |
let hub = hub(HubLimits {
|
|
397 |
+ |
max_connections_per_user: 1,
|
|
398 |
+ |
..HubLimits::default()
|
|
399 |
+ |
});
|
|
400 |
+ |
let r = room();
|
|
401 |
+ |
let u = user();
|
|
402 |
+ |
|
|
403 |
+ |
let held = hub.subscribe(r, u).unwrap();
|
|
404 |
+ |
for _ in 0..10 {
|
|
405 |
+ |
assert!(hub.subscribe(r, u).is_err());
|
|
406 |
+ |
}
|
|
407 |
+ |
// The global counter must not have drifted upward on the rejected path.
|
|
408 |
+ |
assert_eq!(hub.connection_count(), 1);
|
|
409 |
+ |
drop(held);
|
|
410 |
+ |
assert_eq!(hub.connection_count(), 0);
|
|
411 |
+ |
}
|
|
412 |
+ |
|
|
413 |
+ |
#[tokio::test]
|
|
414 |
+ |
async fn a_listener_that_falls_behind_gets_a_gap_not_silence() {
|
|
415 |
+ |
let hub = hub(HubLimits {
|
|
416 |
+ |
room_buffer: 2,
|
|
417 |
+ |
..HubLimits::default()
|
|
418 |
+ |
});
|
|
419 |
+ |
let r = room();
|
|
420 |
+ |
let mut slow = hub.subscribe(r, user()).unwrap();
|
|
421 |
+ |
|
|
422 |
+ |
for i in 0..10 {
|
|
423 |
+ |
hub.publish(r, delete(i));
|
|
424 |
+ |
}
|
|
425 |
+ |
|
|
426 |
+ |
assert_eq!(slow.recv().await, Some(ChatEvent::Gap));
|
|
427 |
+ |
// and the stream keeps working afterward
|
|
428 |
+ |
hub.publish(r, delete(99));
|
|
429 |
+ |
let next = slow.recv().await;
|
|
430 |
+ |
assert!(next.is_some(), "stream survives a gap");
|
|
431 |
+ |
}
|
|
432 |
+ |
|
|
433 |
+ |
#[test]
|
|
434 |
+ |
fn counters_return_to_zero_when_everything_drops() {
|
|
435 |
+ |
let hub = hub(HubLimits::default());
|
|
436 |
+ |
let r = room();
|
|
437 |
+ |
let subs: Vec<_> = (0..5).map(|_| hub.subscribe(r, user()).unwrap()).collect();
|
|
438 |
+ |
assert_eq!(hub.connection_count(), 5);
|
|
439 |
+ |
assert_eq!(hub.room_connection_count(r), 5);
|
|
440 |
+ |
|
|
441 |
+ |
drop(subs);
|
|
442 |
+ |
assert_eq!(hub.connection_count(), 0);
|
|
443 |
+ |
assert_eq!(hub.room_count(), 0);
|
|
444 |
+ |
assert_eq!(hub.room_connection_count(r), 0);
|
|
445 |
+ |
}
|
|
446 |
+ |
}
|