socketioxide/
operators.rs

1//! Operators are used to select sockets to send a packet to,
2//! or to configure the packet that will be emitted.
3//!
4//! They use the builder pattern to chain operators.
5//!
6//! There are two types of operators:
7//! * [`ConfOperators`]: Chainable operators to configure the message to be sent.
8//! * [`BroadcastOperators`]: Chainable operators to select sockets to send a message to and to configure the message to be sent.
9use std::{fmt, future::Future, sync::Arc, time::Duration};
10
11use serde::Serialize;
12use socketioxide_core::Sid;
13
14use crate::{
15    BroadcastError, EmitWithAckError, SendError,
16    ack::{AckInnerStream, AckStream},
17    adapter::{Adapter, LocalAdapter},
18    extract::SocketRef,
19    ns::Namespace,
20    parser::Parser,
21    socket::{RemoteSocket, Socket},
22};
23
24use socketioxide_core::{
25    adapter::{BroadcastFlags, BroadcastOptions, Room, RoomParam},
26    packet::Packet,
27    parser::{Parse, ParserError},
28};
29
30/// Chainable operators to configure the message to be sent.
31pub struct ConfOperators<'a, A: Adapter = LocalAdapter> {
32    timeout: Option<Duration>,
33    volatile: bool,
34    socket: &'a Socket<A>,
35}
36/// Chainable operators to select sockets to send a message to and to configure the message to be sent.
37pub struct BroadcastOperators<A: Adapter = LocalAdapter> {
38    timeout: Option<Duration>,
39    ns: Arc<Namespace<A>>,
40    parser: Parser,
41    opts: BroadcastOptions,
42}
43
44impl<A: Adapter> From<ConfOperators<'_, A>> for BroadcastOperators<A> {
45    fn from(conf: ConfOperators<'_, A>) -> Self {
46        let mut opts = BroadcastOptions::new(conf.socket.id);
47        if conf.volatile {
48            opts.add_flag(BroadcastFlags::Volatile);
49        }
50        Self {
51            timeout: conf.timeout,
52            ns: conf.socket.ns.clone(),
53            parser: conf.socket.parser,
54            opts,
55        }
56    }
57}
58
59// ==== impl ConfOperators operations ====
60impl<'a, A: Adapter> ConfOperators<'a, A> {
61    pub(crate) fn new(sender: &'a Socket<A>) -> Self {
62        Self {
63            timeout: None,
64            volatile: false,
65            socket: sender,
66        }
67    }
68
69    #[doc = include_str!("../docs/operators/to.md")]
70    pub fn to(self, rooms: impl RoomParam) -> BroadcastOperators<A> {
71        BroadcastOperators::from(self).to(rooms)
72    }
73
74    #[doc = include_str!("../docs/operators/within.md")]
75    pub fn within(self, rooms: impl RoomParam) -> BroadcastOperators<A> {
76        BroadcastOperators::from(self).within(rooms)
77    }
78
79    #[doc = include_str!("../docs/operators/except.md")]
80    pub fn except(self, rooms: impl RoomParam) -> BroadcastOperators<A> {
81        BroadcastOperators::from(self).except(rooms)
82    }
83
84    #[doc = include_str!("../docs/operators/local.md")]
85    pub fn local(self) -> BroadcastOperators<A> {
86        BroadcastOperators::from(self).local()
87    }
88
89    #[doc = include_str!("../docs/operators/broadcast.md")]
90    pub fn broadcast(self) -> BroadcastOperators<A> {
91        BroadcastOperators::from(self).broadcast()
92    }
93
94    #[doc = include_str!("../docs/operators/timeout.md")]
95    pub fn timeout(mut self, timeout: Duration) -> Self {
96        self.timeout = Some(timeout);
97        self
98    }
99
100    #[doc = include_str!("../docs/operators/volatile.md")]
101    pub fn volatile(mut self) -> Self {
102        self.volatile = true;
103        self
104    }
105}
106
107// ==== impl ConfOperators consume fns ====
108impl<A: Adapter> ConfOperators<'_, A> {
109    #[doc = include_str!("../docs/operators/emit.md")]
110    pub fn emit<T: ?Sized + Serialize>(
111        mut self,
112        event: impl AsRef<str>,
113        data: &T,
114    ) -> Result<(), SendError> {
115        use crate::SocketError;
116        use crate::socket::PermitExt;
117        if !self.socket.connected() {
118            return Err(SendError::Socket(SocketError::Closed));
119        }
120
121        if self.volatile {
122            let packet = self.get_packet(event, data)?;
123            self.socket
124                .send_raw_volatile(self.socket.parser.encode(packet));
125            return Ok(());
126        }
127
128        let permit = match self.socket.reserve() {
129            Ok(permit) => permit,
130            Err(e) => {
131                #[cfg(feature = "tracing")]
132                tracing::debug!("sending error during emit message: {e:?}");
133                return Err(SendError::Socket(e));
134            }
135        };
136        let packet = self.get_packet(event, data)?;
137        permit.send(packet, self.socket.parser);
138
139        Ok(())
140    }
141
142    #[doc = include_str!("../docs/operators/emit_with_ack.md")]
143    pub fn emit_with_ack<T: ?Sized + Serialize, V>(
144        mut self,
145        event: impl AsRef<str>,
146        data: &T,
147    ) -> Result<AckStream<V>, SendError> {
148        use crate::SocketError;
149        if !self.socket.connected() {
150            return Err(SendError::Socket(SocketError::Closed));
151        }
152        let permit = match self.socket.reserve() {
153            Ok(permit) => permit,
154            Err(e) => {
155                #[cfg(feature = "tracing")]
156                tracing::debug!("sending error during emit message: {e:?}");
157                return Err(SendError::Socket(e));
158            }
159        };
160        let timeout = self
161            .timeout
162            .unwrap_or_else(|| self.socket.get_io().config().ack_timeout);
163        let packet = self.get_packet(event, data)?;
164        let rx = self.socket.send_with_ack_permit(packet, permit);
165        let stream = AckInnerStream::send(rx, timeout, self.socket.id);
166        Ok(AckStream::<V>::new(stream, self.socket.parser))
167    }
168
169    #[doc = include_str!("../docs/operators/join.md")]
170    pub fn join(self, rooms: impl RoomParam) {
171        self.socket.join(rooms)
172    }
173
174    #[doc = include_str!("../docs/operators/leave.md")]
175    pub async fn leave(self, rooms: impl RoomParam) {
176        self.socket.leave(rooms)
177    }
178
179    /// Creates a packet with the given event and data.
180    fn get_packet<T: ?Sized + Serialize>(
181        &mut self,
182        event: impl AsRef<str>,
183        data: &T,
184    ) -> Result<Packet, ParserError> {
185        let ns = self.socket.ns.path.clone();
186        let event = event.as_ref();
187        let data = self.socket.parser.encode_value(&data, Some(event))?;
188        Ok(Packet::event(ns, data))
189    }
190}
191
192impl<A: Adapter> BroadcastOperators<A> {
193    pub(crate) fn new(ns: Arc<Namespace<A>>, parser: Parser) -> Self {
194        Self {
195            timeout: None,
196            ns,
197            parser,
198            opts: BroadcastOptions::default(),
199        }
200    }
201    pub(crate) fn from_sock(ns: Arc<Namespace<A>>, sid: Sid, parser: Parser) -> Self {
202        Self {
203            timeout: None,
204            ns,
205            parser,
206            opts: BroadcastOptions::new(sid),
207        }
208    }
209
210    /// Returns the namespace path of the broadcast operators.
211    pub fn ns_path(&self) -> &str {
212        &self.ns.path
213    }
214
215    #[doc = include_str!("../docs/operators/to.md")]
216    pub fn to(mut self, rooms: impl RoomParam) -> Self {
217        self.opts.rooms.extend(rooms.into_room_iter());
218        self.broadcast()
219    }
220
221    #[doc = include_str!("../docs/operators/within.md")]
222    pub fn within(mut self, rooms: impl RoomParam) -> Self {
223        self.opts.rooms.extend(rooms.into_room_iter());
224        self
225    }
226
227    #[doc = include_str!("../docs/operators/except.md")]
228    pub fn except(mut self, rooms: impl RoomParam) -> Self {
229        self.opts.except.extend(rooms.into_room_iter());
230        self.broadcast()
231    }
232
233    #[doc = include_str!("../docs/operators/local.md")]
234    pub fn local(mut self) -> Self {
235        self.opts.add_flag(BroadcastFlags::Local);
236        self
237    }
238
239    #[doc = include_str!("../docs/operators/broadcast.md")]
240    pub fn broadcast(mut self) -> Self {
241        self.opts.add_flag(BroadcastFlags::Broadcast);
242        self
243    }
244
245    #[doc = include_str!("../docs/operators/timeout.md")]
246    pub fn timeout(mut self, timeout: Duration) -> Self {
247        self.timeout = Some(timeout);
248        self
249    }
250
251    #[doc = include_str!("../docs/operators/volatile.md")]
252    pub fn volatile(mut self) -> Self {
253        self.opts.add_flag(BroadcastFlags::Volatile);
254        self
255    }
256}
257
258// ==== impl BroadcastOperators consume fns ====
259impl<A: Adapter> BroadcastOperators<A> {
260    #[doc = include_str!("../docs/operators/emit.md")]
261    pub fn emit<T: ?Sized + Serialize>(
262        mut self,
263        event: impl AsRef<str>,
264        data: &T,
265    ) -> impl Future<Output = Result<(), BroadcastError>> + Send {
266        let packet = self.get_packet(event, data);
267        async move {
268            self.ns
269                .adapter
270                .broadcast(packet?, self.opts)
271                .await
272                .inspect_err(|_e| {
273                    #[cfg(feature = "tracing")]
274                    tracing::debug!("broadcast error: {_e}");
275                })?;
276            Ok(())
277        }
278    }
279
280    #[doc = include_str!("../docs/operators/emit_with_ack.md")]
281    pub fn emit_with_ack<T: ?Sized + Serialize, V>(
282        mut self,
283        event: impl AsRef<str>,
284        data: &T,
285    ) -> impl Future<Output = Result<AckStream<V, A>, EmitWithAckError>> + Send {
286        let packet = self.get_packet(event, data);
287        async move {
288            let stream = self
289                .ns
290                .adapter
291                .broadcast_with_ack(packet?, self.opts, self.timeout)
292                .await
293                .map_err(|e| EmitWithAckError::Adapter(Box::new(e)))?;
294            Ok(AckStream::new(stream, self.parser))
295        }
296    }
297
298    #[doc = include_str!("../docs/operators/sockets.md")]
299    pub fn sockets(self) -> Vec<SocketRef<A>> {
300        let ids = self.ns.adapter.get_local().sockets(self.opts);
301
302        ids.into_iter()
303            .filter_map(|id| self.ns.get_socket(id).ok())
304            .map(SocketRef::from)
305            .collect()
306    }
307
308    #[doc = include_str!("../docs/operators/fetch_sockets.md")]
309    pub async fn fetch_sockets(self) -> Result<Vec<RemoteSocket<A>>, A::Error> {
310        let sockets = self
311            .ns
312            .adapter
313            .fetch_sockets(self.opts)
314            .await?
315            .into_iter()
316            .map(|data| RemoteSocket::new(data, &self.ns.adapter, self.parser))
317            .collect();
318        Ok(sockets)
319    }
320
321    #[doc = include_str!("../docs/operators/disconnect.md")]
322    pub async fn disconnect(self) -> Result<(), BroadcastError> {
323        self.ns.adapter.disconnect_socket(self.opts).await
324    }
325
326    #[doc = include_str!("../docs/operators/join.md")]
327    #[allow(clippy::manual_async_fn)] // related to issue: https://github.com/rust-lang/rust-clippy/issues/12664
328    pub fn join(self, rooms: impl RoomParam) -> impl Future<Output = Result<(), A::Error>> + Send {
329        async move { self.ns.adapter.add_sockets(self.opts, rooms).await }
330    }
331
332    #[doc = include_str!("../docs/operators/leave.md")]
333    #[allow(clippy::manual_async_fn)] // related to issue: https://github.com/rust-lang/rust-clippy/issues/12664
334    pub fn leave(self, rooms: impl RoomParam) -> impl Future<Output = Result<(), A::Error>> + Send {
335        async move { self.ns.adapter.del_sockets(self.opts, rooms).await }
336    }
337
338    #[doc = include_str!("../docs/operators/rooms.md")]
339    pub async fn rooms(self) -> Result<Vec<Room>, A::Error> {
340        self.ns.adapter.rooms(self.opts).await
341    }
342
343    #[doc = include_str!("../docs/operators/get_socket.md")]
344    pub fn get_socket(&self, sid: Sid) -> Option<SocketRef<A>> {
345        self.ns.get_socket(sid).map(SocketRef::from).ok()
346    }
347
348    /// Creates a packet with the given event and data.
349    fn get_packet<T: ?Sized + Serialize>(
350        &mut self,
351        event: impl AsRef<str>,
352        data: &T,
353    ) -> Result<Packet, ParserError> {
354        let ns = self.ns.path.clone();
355        let data = self.parser.encode_value(data, Some(event.as_ref()))?;
356        Ok(Packet::event(ns, data))
357    }
358}
359
360impl<'a, A: Adapter> Clone for ConfOperators<'a, A> {
361    fn clone(&self) -> Self {
362        Self {
363            timeout: self.timeout,
364            volatile: self.volatile,
365            socket: self.socket,
366        }
367    }
368}
369impl<A: Adapter> Clone for BroadcastOperators<A> {
370    fn clone(&self) -> Self {
371        Self {
372            ns: self.ns.clone(),
373            opts: self.opts.clone(),
374            timeout: self.timeout,
375            parser: self.parser,
376        }
377    }
378}
379impl<A: Adapter> fmt::Debug for BroadcastOperators<A> {
380    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
381        f.debug_struct("BroadcastOperators")
382            .field("ns", &self.ns)
383            .field("opts", &self.opts)
384            .field("timeout", &self.timeout)
385            .finish()
386    }
387}
388
389impl<A: Adapter> fmt::Debug for ConfOperators<'_, A> {
390    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
391        f.debug_struct("ConfOperators")
392            .field("timeout", &self.timeout)
393            .finish()
394    }
395}
socketioxide/operators.rs

socketioxide/
operators.rs
