rainier_broadcast/
manager.rs

1//! The thing an application holds — [`Broadcasting`].
2
3use std::sync::Arc;
4
5use serde_json::Value;
6
7use rainier_support::Result;
8
9use crate::broadcaster::{Broadcaster, LogBroadcaster};
10use crate::channel::Channel;
11use crate::event::{Broadcast, Broadcastable};
12
13/// Publishes broadcasts through the configured driver.
14///
15/// What the `Broadcast` facade
16/// resolves to.
17pub struct Broadcasting {
18    driver: Arc<dyn Broadcaster>,
19}
20
21impl Broadcasting {
22    /// Publishing through `driver`.
23    pub fn new(driver: Arc<dyn Broadcaster>) -> Self {
24        Self { driver }
25    }
26
27    /// Publishing to the log — the safe default, and what an unconfigured
28    /// application gets.
29    pub fn log() -> Self {
30        Self::new(Arc::new(LogBroadcaster))
31    }
32
33    /// The driver's name.
34    pub fn driver_name(&self) -> &'static str {
35        self.driver.name()
36    }
37
38    /// The driver, for the auth endpoint's signature.
39    pub fn driver(&self) -> &Arc<dyn Broadcaster> {
40        &self.driver
41    }
42
43    /// Publish an event.
44    ///
45    /// Does nothing when the event says [`broadcast_when`](Broadcastable::broadcast_when)
46    /// is false, or when it named no channels — both are answers, not errors.
47    pub async fn event<E: Broadcastable + ?Sized>(&self, event: &E) -> Result<()> {
48        if !event.broadcast_when() {
49            return Ok(());
50        }
51        self.send(Broadcast::of(event)?).await
52    }
53
54    /// Publish an event to everyone **except** one socket.
55    ///
56    /// The browser that caused the change has already applied it locally;
57    /// echoing it back makes its own edit flicker. `socket_id` comes from the
58    /// request's `X-Socket-ID` header, and `None` means broadcast to all.
59    pub async fn event_except<E: Broadcastable + ?Sized>(
60        &self,
61        event: &E,
62        socket_id: Option<String>,
63    ) -> Result<()> {
64        if !event.broadcast_when() {
65            return Ok(());
66        }
67        self.send(Broadcast::of(event)?.except_maybe(socket_id)).await
68    }
69
70    /// Publish a payload directly, without an event type.
71    ///
72    /// For something assembled at run time — a channel name computed from
73    /// data, a payload proxied from elsewhere.
74    pub async fn to(
75        &self,
76        channels: Vec<Channel>,
77        event: impl Into<String>,
78        payload: Value,
79    ) -> Result<()> {
80        self.send(Broadcast::new(event, channels, payload)).await
81    }
82
83    /// Publish an already-rendered broadcast.
84    pub async fn send(&self, broadcast: Broadcast) -> Result<()> {
85        // No channels is not a failure — an event whose `broadcast_on`
86        // returned nothing has decided this instance is not visible to anyone.
87        if broadcast.is_empty() {
88            return Ok(());
89        }
90        self.driver.publish(&broadcast).await
91    }
92}
93
94impl std::fmt::Debug for Broadcasting {
95    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96        f.debug_struct("Broadcasting").field("driver", &self.driver.name()).finish()
97    }
98}
99
100#[cfg(test)]
101mod tests {
102    use super::*;
103    use crate::broadcaster::MemoryBroadcaster;
104    use serde::Serialize;
105
106    #[derive(Serialize)]
107    struct Ping {
108        room: u64,
109        #[serde(skip)]
110        visible: bool,
111    }
112
113    impl Broadcastable for Ping {
114        fn broadcast_on(&self) -> Vec<Channel> {
115            vec![Channel::private(format!("room.{}", self.room))]
116        }
117        fn broadcast_when(&self) -> bool {
118            self.visible
119        }
120    }
121
122    fn manager() -> (Broadcasting, Arc<MemoryBroadcaster>) {
123        let driver = Arc::new(MemoryBroadcaster::new());
124        (Broadcasting::new(driver.clone()), driver)
125    }
126
127    #[tokio::test]
128    async fn an_event_reaches_the_driver() {
129        let (broadcasting, driver) = manager();
130
131        broadcasting.event(&Ping { room: 1, visible: true }).await.unwrap();
132
133        driver.assert_broadcast("Ping", "private-room.1");
134    }
135
136    #[tokio::test]
137    async fn broadcast_when_false_sends_nothing() {
138        let (broadcasting, driver) = manager();
139
140        broadcasting.event(&Ping { room: 1, visible: false }).await.unwrap();
141
142        driver.assert_nothing_broadcast();
143    }
144
145    #[tokio::test]
146    async fn to_others_carries_the_socket_to_skip() {
147        let (broadcasting, driver) = manager();
148
149        broadcasting
150            .event_except(&Ping { room: 1, visible: true }, Some("1234.5678".into()))
151            .await
152            .unwrap();
153
154        assert_eq!(driver.sent()[0].except.as_deref(), Some("1234.5678"));
155    }
156
157    #[tokio::test]
158    async fn a_request_with_no_socket_header_broadcasts_to_everyone() {
159        let (broadcasting, driver) = manager();
160
161        broadcasting.event_except(&Ping { room: 1, visible: true }, None).await.unwrap();
162
163        assert_eq!(driver.sent()[0].except, None);
164    }
165
166    #[tokio::test]
167    async fn an_event_with_no_channels_never_reaches_the_driver() {
168        let (broadcasting, driver) = manager();
169
170        broadcasting.to(vec![], "manual", serde_json::json!({ "a": 1 })).await.unwrap();
171
172        driver.assert_nothing_broadcast();
173    }
174}
rainier_broadcast/manager.rs

rainier_broadcast/
manager.rs
