steamkit/networking/steam3/
i_connection.rs

1//! The connection trait and its events.
2//!
3//! [`IConnection`] is implemented by the transports (TCP, UDP, WebSocket). Their
4//! events come through [`ConnectionEvents`], in two forms:
5//!
6//! * [`ConnectionEvents::events`] is one ordered, lossless stream of
7//!   [`ConnectionEvent`]s. Anything protocol-related should use this, and the
8//!   crate itself only uses this.
9//! * Three per-event `broadcast` receivers, for callers that only care about one
10//!   kind of event. A receiver that falls more than [`CONNECTION_EVENT_CAPACITY`]
11//!   events behind gets `RecvError::Lagged` and loses events, and the three
12//!   channels aren't ordered relative to each other.
13
14use std::net::{IpAddr, SocketAddr};
15use std::sync::Arc;
16use std::time::Duration;
17
18use async_trait::async_trait;
19use bytes::Bytes;
20use parking_lot::Mutex;
21use tokio::sync::{broadcast, mpsc};
22
23use crate::error::SteamError;
24
25use super::disconnected_event_args::DisconnectedEventArgs;
26use super::net_msg_event_args::NetMsgEventArgs;
27use super::protocol_type::ProtocolTypes;
28
29/// Capacity of each per-event broadcast channel. The ordered stream from
30/// [`ConnectionEvents::events`] is unbounded.
31pub const CONNECTION_EVENT_CAPACITY: usize = 64;
32
33/// A transport connection to a CM server.
34#[async_trait]
35pub trait IConnection: Send + Sync {
36    /// The remote endpoint of the current connection.
37    ///
38    /// This is `Some` between the `Connected` and `Disconnected` events,
39    /// inclusive.
40    fn current_end_point(&self) -> Option<SocketAddr>;
41
42    /// The type of communication protocol that this connection uses.
43    fn protocol_types(&self) -> ProtocolTypes;
44
45    /// Connects to the specified endpoint.
46    async fn connect(&self, end_point: SocketAddr, timeout: Duration) -> Result<(), SteamError>;
47
48    /// Disconnects. `user_initiated` is `true` when the caller asked for it.
49    async fn disconnect(&self, user_initiated: bool);
50
51    /// Sends the specified data packet.
52    async fn send(&self, data: Bytes) -> Result<(), SteamError>;
53
54    /// The local IP of the underlying socket, when one is bound.
55    fn local_ip(&self) -> Option<IpAddr>;
56}
57
58/// One event raised by a connection.
59#[derive(Clone, Debug, PartialEq, Eq)]
60pub enum ConnectionEvent {
61    /// A message was received.
62    NetMsgReceived(NetMsgEventArgs),
63    /// The connection is established.
64    Connected,
65    /// The connection is gone.
66    Disconnected(DisconnectedEventArgs),
67}
68
69/// The events of a connection. Every [`IConnection`] in this crate implements
70/// this too.
71pub trait ConnectionEvents: Send + Sync {
72    /// Subscribes to all three events as one stream.
73    ///
74    /// The stream is unbounded, so no event is ever dropped, and it delivers the
75    /// events in the order the connection raised them. Subscribe before calling
76    /// `connect`, or the first events are missed.
77    fn events(&self) -> mpsc::UnboundedReceiver<ConnectionEvent>;
78
79    /// Subscribes to the `NetMsgReceived` event.
80    fn net_msg_received(&self) -> broadcast::Receiver<NetMsgEventArgs>;
81
82    /// Subscribes to the `Connected` event.
83    fn connected(&self) -> broadcast::Receiver<()>;
84
85    /// Subscribes to the `Disconnected` event.
86    fn disconnected(&self) -> broadcast::Receiver<DisconnectedEventArgs>;
87}
88
89/// A connection that also publishes its events.
90/// [`EnvelopeEncryptedConnection`](super::envelope_encrypted_connection::EnvelopeEncryptedConnection)
91/// takes its inner connection as an `Arc<dyn Connection>`.
92pub trait Connection: IConnection + ConnectionEvents {}
93
94impl<T> Connection for T where T: IConnection + ConnectionEvents + ?Sized {}
95
96/// The sending halves of the connection event channels, used by every
97/// connection type. Clones share the same channels.
98#[derive(Clone, Debug)]
99pub struct ConnectionEventSenders {
100    net_msg_received: broadcast::Sender<NetMsgEventArgs>,
101    connected: broadcast::Sender<()>,
102    disconnected: broadcast::Sender<DisconnectedEventArgs>,
103    /// Subscribers of the ordered stream. The lock is held while an event is
104    /// handed to all of them, which gives events raised from different tasks
105    /// one order.
106    subscribers: Arc<Mutex<Vec<mpsc::UnboundedSender<ConnectionEvent>>>>,
107}
108
109impl ConnectionEventSenders {
110    /// Creates the event channels; `capacity` is the size of each broadcast channel.
111    pub fn new(capacity: usize) -> Self {
112        let (net_msg_received, _) = broadcast::channel(capacity);
113        let (connected, _) = broadcast::channel(capacity);
114        let (disconnected, _) = broadcast::channel(capacity);
115
116        Self {
117            net_msg_received,
118            connected,
119            disconnected,
120            subscribers: Arc::new(Mutex::new(Vec::new())),
121        }
122    }
123
124    /// Creates a new set of event channels with [`CONNECTION_EVENT_CAPACITY`].
125    pub fn with_default_capacity() -> Self {
126        Self::new(CONNECTION_EVENT_CAPACITY)
127    }
128
129    /// Subscribes to the ordered, lossless stream of all three events.
130    pub fn subscribe_events(&self) -> mpsc::UnboundedReceiver<ConnectionEvent> {
131        let (sender, receiver) = mpsc::unbounded_channel();
132        self.subscribers.lock().push(sender);
133        receiver
134    }
135
136    /// The number of live subscribers of the ordered stream.
137    pub fn event_subscriber_count(&self) -> usize {
138        let mut subscribers = self.subscribers.lock();
139        subscribers.retain(|subscriber| !subscriber.is_closed());
140        subscribers.len()
141    }
142
143    /// Subscribes to the `NetMsgReceived` event.
144    pub fn subscribe_net_msg_received(&self) -> broadcast::Receiver<NetMsgEventArgs> {
145        self.net_msg_received.subscribe()
146    }
147
148    /// Subscribes to the `Connected` event.
149    pub fn subscribe_connected(&self) -> broadcast::Receiver<()> {
150        self.connected.subscribe()
151    }
152
153    /// Subscribes to the `Disconnected` event.
154    pub fn subscribe_disconnected(&self) -> broadcast::Receiver<DisconnectedEventArgs> {
155        self.disconnected.subscribe()
156    }
157
158    /// Raises the `NetMsgReceived` event. It's dropped if nobody is subscribed.
159    pub fn emit_net_msg_received(&self, args: NetMsgEventArgs) {
160        let mut subscribers = self.subscribers.lock();
161        if self.net_msg_received.receiver_count() > 0 {
162            let _ = self.net_msg_received.send(args.clone());
163        }
164        publish(&mut subscribers, ConnectionEvent::NetMsgReceived(args));
165    }
166
167    /// Raises the `Connected` event.
168    pub fn emit_connected(&self) {
169        let mut subscribers = self.subscribers.lock();
170        let _ = self.connected.send(());
171        publish(&mut subscribers, ConnectionEvent::Connected);
172    }
173
174    /// Raises the `Disconnected` event.
175    pub fn emit_disconnected(&self, args: DisconnectedEventArgs) {
176        let mut subscribers = self.subscribers.lock();
177        let _ = self.disconnected.send(args);
178        publish(&mut subscribers, ConnectionEvent::Disconnected(args));
179    }
180}
181
182/// Hands an event to every subscriber of the ordered stream and drops the ones
183/// that have gone away.
184fn publish(subscribers: &mut Vec<mpsc::UnboundedSender<ConnectionEvent>>, event: ConnectionEvent) {
185    if let [only] = subscribers.as_slice() {
186        // Common case: one subscriber, so no need to clone the event.
187        if only.send(event).is_err() {
188            subscribers.clear();
189        }
190        return;
191    }
192
193    subscribers.retain(|subscriber| subscriber.send(event.clone()).is_ok());
194}
195
196impl Default for ConnectionEventSenders {
197    fn default() -> Self {
198        Self::with_default_capacity()
199    }
200}
steamkit/networking/steam3/i_connection.rs

steamkit/networking/steam3/
i_connection.rs
