netcode/
client.rs

1//! The netcode client.
2
3use std::collections::VecDeque;
4use std::fmt;
5use std::net::{SocketAddr, UdpSocket};
6
7use log::{debug, info};
8
9use crate::packet::{AllowedPackets, Packet};
10use crate::replay::ReplayProtection;
11use crate::token::{self, CHALLENGE_TOKEN_BYTES, ConnectToken};
12use crate::{
13    CONNECT_TOKEN_BYTES, Error, KEY_BYTES, Key, MAX_PACKET_BYTES, MAX_PAYLOAD_BYTES,
14    NUM_DISCONNECT_PACKETS, PACKET_QUEUE_SIZE, PACKET_SEND_RATE, socket,
15};
16
17/// The state of a [`Client`].
18///
19/// The initial state is [`Disconnected`](ClientState::Disconnected); the goal state is
20/// [`Connected`](ClientState::Connected). The [`is_error`](ClientState::is_error),
21/// [`is_connecting`](ClientState::is_connecting) and
22/// [`is_disconnected`](ClientState::is_disconnected) predicates classify the rest.
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub enum ClientState {
25    /// The connect token expired before the client finished connecting.
26    ConnectTokenExpired,
27    /// The connect token passed to [`Client::connect`] was invalid.
28    InvalidConnectToken,
29    /// The established connection timed out.
30    ConnectionTimedOut,
31    /// The server stopped responding while the client was sending connection responses.
32    ConnectionResponseTimedOut,
33    /// The server never responded to the client's connection requests.
34    ConnectionRequestTimedOut,
35    /// The server denied the connection (for example, because it was full).
36    ConnectionDenied,
37    /// Not connected. The initial state.
38    Disconnected,
39    /// Sending connection request packets to the server.
40    SendingConnectionRequest,
41    /// Sending connection response packets to the server.
42    SendingConnectionResponse,
43    /// Connected: payload packets can be exchanged with the server.
44    Connected,
45}
46
47impl ClientState {
48    /// Whether this is one of the error states.
49    pub fn is_error(self) -> bool {
50        matches!(
51            self,
52            ClientState::ConnectTokenExpired
53                | ClientState::InvalidConnectToken
54                | ClientState::ConnectionTimedOut
55                | ClientState::ConnectionResponseTimedOut
56                | ClientState::ConnectionRequestTimedOut
57                | ClientState::ConnectionDenied
58        )
59    }
60
61    /// Whether the client is partway through connecting: sending connection requests
62    /// or connection responses.
63    pub fn is_connecting(self) -> bool {
64        matches!(
65            self,
66            ClientState::SendingConnectionRequest | ClientState::SendingConnectionResponse
67        )
68    }
69
70    /// Whether the client is disconnected, either cleanly
71    /// ([`Disconnected`](ClientState::Disconnected)) or in an error state.
72    pub fn is_disconnected(self) -> bool {
73        self == ClientState::Disconnected || self.is_error()
74    }
75}
76
77impl fmt::Display for ClientState {
78    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
79        let name = match self {
80            ClientState::ConnectTokenExpired => "connect token expired",
81            ClientState::InvalidConnectToken => "invalid connect token",
82            ClientState::ConnectionTimedOut => "connection timed out",
83            ClientState::ConnectionResponseTimedOut => "connection response timed out",
84            ClientState::ConnectionRequestTimedOut => "connection request timed out",
85            ClientState::ConnectionDenied => "connection denied",
86            ClientState::Disconnected => "disconnected",
87            ClientState::SendingConnectionRequest => "sending connection request",
88            ClientState::SendingConnectionResponse => "sending connection response",
89            ClientState::Connected => "connected",
90        };
91        f.write_str(name)
92    }
93}
94
95/// A netcode client.
96///
97/// Drive it by calling [`update`](Client::update) once per frame with the current
98/// time. All methods must be called from one thread; the client performs no internal
99/// synchronization.
100pub struct Client {
101    state: ClientState,
102    time: f64,
103    connect_start_time: f64,
104    last_packet_send_time: f64,
105    last_packet_receive_time: f64,
106    should_disconnect: Option<ClientState>,
107    sequence: u64,
108    client_index: usize,
109    max_clients: usize,
110    server_address_index: usize,
111    server_address: Option<SocketAddr>,
112    connect_token: Option<ConnectToken>,
113    socket_ipv4: Option<UdpSocket>,
114    socket_ipv6: Option<UdpSocket>,
115    primary_is_ipv4: bool,
116    write_packet_key: Key,
117    read_packet_key: Key,
118    replay_protection: ReplayProtection,
119    packet_receive_queue: VecDeque<(Vec<u8>, u64)>,
120    challenge_token_sequence: u64,
121    challenge_token_data: [u8; CHALLENGE_TOKEN_BYTES],
122}
123
124impl Client {
125    /// Creates a client with a socket bound to `bind_address`. Bind to port 0 to let
126    /// the operating system pick a port.
127    ///
128    /// The client can only reach server addresses in the same address family as the
129    /// bind address; use [`new_dual`](Client::new_dual) to support connect tokens
130    /// that mix IPv4 and IPv6 server addresses.
131    pub fn new(bind_address: SocketAddr, time: f64) -> Result<Self, Error> {
132        match bind_address {
133            SocketAddr::V4(_) => Self::create(Some(bind_address), None, true, time),
134            SocketAddr::V6(_) => Self::create(None, Some(bind_address), false, time),
135        }
136    }
137
138    /// Creates a client with one IPv4 socket and one IPv6 socket, so it can connect
139    /// to server addresses of either family.
140    pub fn new_dual(
141        bind_address_ipv4: SocketAddr,
142        bind_address_ipv6: SocketAddr,
143        time: f64,
144    ) -> Result<Self, Error> {
145        if !bind_address_ipv4.is_ipv4() || !bind_address_ipv6.is_ipv6() {
146            return Err(Error::Io(std::io::Error::new(
147                std::io::ErrorKind::InvalidInput,
148                "new_dual requires one IPv4 and one IPv6 bind address",
149            )));
150        }
151        Self::create(Some(bind_address_ipv4), Some(bind_address_ipv6), true, time)
152    }
153
154    fn create(
155        bind_address_ipv4: Option<SocketAddr>,
156        bind_address_ipv6: Option<SocketAddr>,
157        primary_is_ipv4: bool,
158        time: f64,
159    ) -> Result<Self, Error> {
160        let socket_ipv4 = bind_address_ipv4.map(socket::create_socket).transpose()?;
161        let socket_ipv6 = bind_address_ipv6.map(socket::create_socket).transpose()?;
162
163        let client = Self {
164            state: ClientState::Disconnected,
165            time,
166            connect_start_time: 0.0,
167            last_packet_send_time: -1000.0,
168            last_packet_receive_time: -1000.0,
169            should_disconnect: None,
170            sequence: 0,
171            client_index: 0,
172            max_clients: 0,
173            server_address_index: 0,
174            server_address: None,
175            connect_token: None,
176            socket_ipv4,
177            socket_ipv6,
178            primary_is_ipv4,
179            write_packet_key: [0; KEY_BYTES],
180            read_packet_key: [0; KEY_BYTES],
181            replay_protection: ReplayProtection::new(),
182            packet_receive_queue: VecDeque::new(),
183            challenge_token_sequence: 0,
184            challenge_token_data: [0; CHALLENGE_TOKEN_BYTES],
185        };
186
187        info!("client started on port {}", client.port());
188
189        Ok(client)
190    }
191
192    /// Starts connecting to the servers listed in the connect token, trying each
193    /// address in order until one accepts.
194    ///
195    /// Track progress by polling [`state`](Client::state) after each update. On an
196    /// invalid token this returns an error and the client transitions to
197    /// [`ClientState::InvalidConnectToken`].
198    pub fn connect(&mut self, connect_token: &[u8; CONNECT_TOKEN_BYTES]) -> Result<(), Error> {
199        self.disconnect();
200
201        let connect_token = match ConnectToken::read(connect_token) {
202            Ok(connect_token) => connect_token,
203            Err(error) => {
204                self.set_state(ClientState::InvalidConnectToken);
205                return Err(error);
206            }
207        };
208
209        self.server_address_index = 0;
210        self.server_address = Some(connect_token.server_addresses[0]);
211
212        if connect_token.server_addresses.len() == 1 {
213            info!("client connecting to server {}", connect_token.server_addresses[0]);
214        } else {
215            info!(
216                "client connecting to server {} [1/{}]",
217                connect_token.server_addresses[0],
218                connect_token.server_addresses.len()
219            );
220        }
221
222        self.read_packet_key = connect_token.server_to_client_key;
223        self.write_packet_key = connect_token.client_to_server_key;
224        self.connect_token = Some(connect_token);
225
226        self.reset_before_next_connect();
227        self.set_state(ClientState::SendingConnectionRequest);
228
229        Ok(())
230    }
231
232    /// Advances the client to the current time: receives and processes packets, sends
233    /// queued packets and keep-alives, and applies timeouts.
234    pub fn update(&mut self, time: f64) {
235        self.time = time;
236
237        self.receive_packets();
238        self.send_packets();
239
240        if self.state.is_connecting() {
241            let connect_token = self.connect_token.as_ref().unwrap();
242            let expire_seconds = connect_token.expire_timestamp - connect_token.create_timestamp;
243            if self.time - self.connect_start_time >= expire_seconds as f64 {
244                info!("client connect failed. connect token expired");
245                self.disconnect_internal(ClientState::ConnectTokenExpired, false);
246                return;
247            }
248        }
249
250        if let Some(disconnect_state) = self.should_disconnect {
251            debug!("client should disconnect -> {disconnect_state}");
252            if self.connect_to_next_server() {
253                return;
254            }
255            self.disconnect_internal(disconnect_state, false);
256            return;
257        }
258
259        let timeout_seconds =
260            self.connect_token.as_ref().map_or(0, |connect_token| connect_token.timeout_seconds);
261        let timed_out =
262            timeout_seconds > 0 && self.last_packet_receive_time + (timeout_seconds as f64) < time;
263        if !timed_out {
264            return;
265        }
266
267        match self.state {
268            ClientState::SendingConnectionRequest => {
269                info!("client connect failed. connection request timed out");
270                if !self.connect_to_next_server() {
271                    self.disconnect_internal(ClientState::ConnectionRequestTimedOut, false);
272                }
273            }
274            ClientState::SendingConnectionResponse => {
275                info!("client connect failed. connection response timed out");
276                if !self.connect_to_next_server() {
277                    self.disconnect_internal(ClientState::ConnectionResponseTimedOut, false);
278                }
279            }
280            ClientState::Connected => {
281                info!("client connection timed out");
282                self.disconnect_internal(ClientState::ConnectionTimedOut, false);
283            }
284            _ => {}
285        }
286    }
287
288    /// Sends a payload packet to the server. Does nothing unless the client is
289    /// connected.
290    pub fn send_packet(&mut self, payload: &[u8]) -> Result<(), Error> {
291        if payload.is_empty() || payload.len() > MAX_PAYLOAD_BYTES {
292            return Err(Error::InvalidPayloadSize(payload.len()));
293        }
294
295        if self.state != ClientState::Connected {
296            return Ok(());
297        }
298
299        self.send_packet_to_server(&Packet::Payload(payload.to_vec()));
300
301        Ok(())
302    }
303
304    /// Pops the next payload packet received from the server, along with its sequence
305    /// number.
306    pub fn receive_packet(&mut self) -> Option<(Vec<u8>, u64)> {
307        self.packet_receive_queue.pop_front()
308    }
309
310    /// Disconnects from the server, sending redundant disconnect packets so the
311    /// server finds out quickly.
312    pub fn disconnect(&mut self) {
313        self.disconnect_internal(ClientState::Disconnected, true);
314    }
315
316    /// The current client state.
317    pub fn state(&self) -> ClientState {
318        self.state
319    }
320
321    /// The slot this client occupies on the server, valid while connected.
322    pub fn client_index(&self) -> usize {
323        self.client_index
324    }
325
326    /// The number of client slots on the server, valid while connected.
327    pub fn max_clients(&self) -> usize {
328        self.max_clients
329    }
330
331    /// The sequence number of the next packet this client will send.
332    pub fn next_packet_sequence(&self) -> u64 {
333        self.sequence
334    }
335
336    /// The local port the client is bound to.
337    pub fn port(&self) -> u16 {
338        let socket = if self.primary_is_ipv4 {
339            self.socket_ipv4.as_ref()
340        } else {
341            self.socket_ipv6.as_ref()
342        };
343        socket.and_then(|socket| socket.local_addr().ok()).map_or(0, |address| address.port())
344    }
345
346    /// The server address the client is currently connecting or connected to.
347    pub fn server_address(&self) -> Option<SocketAddr> {
348        self.server_address
349    }
350
351    // ----------------------------------------------------------------
352
353    fn set_state(&mut self, state: ClientState) {
354        debug!("client changed state from '{}' to '{}'", self.state, state);
355        self.state = state;
356    }
357
358    fn reset_before_next_connect(&mut self) {
359        self.connect_start_time = self.time;
360        self.last_packet_send_time = self.time - 1.0;
361        self.last_packet_receive_time = self.time;
362        self.should_disconnect = None;
363        self.challenge_token_sequence = 0;
364        self.challenge_token_data = [0; CHALLENGE_TOKEN_BYTES];
365        self.replay_protection.reset();
366    }
367
368    fn reset_connection_data(&mut self, state: ClientState) {
369        self.sequence = 0;
370        self.client_index = 0;
371        self.max_clients = 0;
372        self.connect_start_time = 0.0;
373        self.server_address_index = 0;
374        self.server_address = None;
375        self.connect_token = None;
376        self.write_packet_key = [0; KEY_BYTES];
377        self.read_packet_key = [0; KEY_BYTES];
378        self.set_state(state);
379        self.reset_before_next_connect();
380        self.packet_receive_queue.clear();
381    }
382
383    fn disconnect_internal(
384        &mut self,
385        destination_state: ClientState,
386        send_disconnect_packets: bool,
387    ) {
388        debug_assert!(destination_state.is_disconnected());
389
390        if self.state.is_disconnected() || self.state == destination_state {
391            return;
392        }
393
394        info!("client disconnected");
395
396        if send_disconnect_packets {
397            debug!("client sent disconnect packets to server");
398            for _ in 0..NUM_DISCONNECT_PACKETS {
399                self.send_packet_to_server(&Packet::Disconnect);
400            }
401        }
402
403        self.reset_connection_data(destination_state);
404    }
405
406    fn connect_to_next_server(&mut self) -> bool {
407        let Some(connect_token) = self.connect_token.as_ref() else {
408            return false;
409        };
410        let num_server_addresses = connect_token.server_addresses.len();
411
412        if self.server_address_index + 1 >= num_server_addresses {
413            debug!("client has no more servers to connect to");
414            return false;
415        }
416
417        self.server_address_index += 1;
418        let server_address = connect_token.server_addresses[self.server_address_index];
419        self.server_address = Some(server_address);
420
421        self.reset_before_next_connect();
422
423        info!(
424            "client connecting to next server {} [{}/{}]",
425            server_address,
426            self.server_address_index + 1,
427            num_server_addresses
428        );
429
430        self.set_state(ClientState::SendingConnectionRequest);
431
432        true
433    }
434
435    fn send_packets(&mut self) {
436        if self.last_packet_send_time + 1.0 / PACKET_SEND_RATE >= self.time {
437            return;
438        }
439
440        match self.state {
441            ClientState::SendingConnectionRequest => {
442                debug!("client sent connection request packet to server");
443                let connect_token = self.connect_token.as_ref().unwrap();
444                let packet = Packet::Request {
445                    protocol_id: connect_token.protocol_id,
446                    expire_timestamp: connect_token.expire_timestamp,
447                    nonce: connect_token.nonce,
448                    private_data: connect_token.private_data.clone(),
449                };
450                self.send_packet_to_server(&packet);
451            }
452            ClientState::SendingConnectionResponse => {
453                debug!("client sent connection response packet to server");
454                let packet = Packet::Response {
455                    challenge_token_sequence: self.challenge_token_sequence,
456                    challenge_token_data: self.challenge_token_data,
457                };
458                self.send_packet_to_server(&packet);
459            }
460            ClientState::Connected => {
461                debug!("client sent connection keep alive packet to server");
462                let packet = Packet::KeepAlive { client_index: 0, max_clients: 0 };
463                self.send_packet_to_server(&packet);
464            }
465            _ => {}
466        }
467    }
468
469    fn send_packet_to_server(&mut self, packet: &Packet) {
470        let Some(server_address) = self.server_address else {
471            return;
472        };
473        let protocol_id =
474            self.connect_token.as_ref().map_or(0, |connect_token| connect_token.protocol_id);
475
476        let mut packet_data = [0u8; MAX_PACKET_BYTES];
477        let packet_bytes = match crate::packet::write_packet(
478            packet,
479            &mut packet_data,
480            self.sequence,
481            &self.write_packet_key,
482            protocol_id,
483        ) {
484            Ok(packet_bytes) => packet_bytes,
485            Err(_) => return,
486        };
487        self.sequence += 1;
488
489        let socket = match server_address {
490            SocketAddr::V4(_) => self.socket_ipv4.as_ref(),
491            SocketAddr::V6(_) => self.socket_ipv6.as_ref(),
492        };
493        match socket {
494            Some(socket) => {
495                let _ = socket.send_to(&packet_data[..packet_bytes], server_address);
496            }
497            None => {
498                debug!("client has no socket for server address family: {server_address}");
499            }
500        }
501
502        self.last_packet_send_time = self.time;
503    }
504
505    fn receive_packets(&mut self) {
506        let mut packet_data = [0u8; MAX_PACKET_BYTES];
507        loop {
508            // the client only talks to one server, so receive on the socket matching
509            // the current server address family. the socket borrow is scoped to the
510            // receive call so packet processing can borrow self mutably.
511            let socket = match self.server_address {
512                Some(SocketAddr::V4(_)) => self.socket_ipv4.as_ref(),
513                Some(SocketAddr::V6(_)) => self.socket_ipv6.as_ref(),
514                None => None,
515            };
516            let Some(socket) = socket else {
517                return;
518            };
519            let Some((packet_bytes, from)) = socket::receive_packet(socket, &mut packet_data)
520            else {
521                return;
522            };
523            self.process_packet(from, &mut packet_data[..packet_bytes]);
524        }
525    }
526
527    fn process_packet(&mut self, from: SocketAddr, packet_data: &mut [u8]) {
528        let Some(connect_token) = self.connect_token.as_ref() else {
529            return;
530        };
531
532        let Some((packet, sequence)) = crate::packet::read_packet(
533            packet_data,
534            Some(&self.read_packet_key),
535            connect_token.protocol_id,
536            token::unix_timestamp(),
537            None,
538            AllowedPackets::CLIENT,
539            Some(&mut self.replay_protection),
540        ) else {
541            return;
542        };
543
544        if Some(from) != self.server_address {
545            return;
546        }
547
548        match packet {
549            Packet::Denied => {
550                if self.state == ClientState::SendingConnectionRequest
551                    || self.state == ClientState::SendingConnectionResponse
552                {
553                    self.should_disconnect = Some(ClientState::ConnectionDenied);
554                    self.last_packet_receive_time = self.time;
555                }
556            }
557
558            Packet::Challenge { challenge_token_sequence, challenge_token_data } => {
559                if self.state == ClientState::SendingConnectionRequest {
560                    debug!("client received connection challenge packet from server");
561                    self.challenge_token_sequence = challenge_token_sequence;
562                    self.challenge_token_data = challenge_token_data;
563                    self.last_packet_receive_time = self.time;
564                    self.set_state(ClientState::SendingConnectionResponse);
565                }
566            }
567
568            Packet::KeepAlive { client_index, max_clients } => match self.state {
569                ClientState::Connected => {
570                    debug!("client received connection keep alive packet from server");
571                    self.last_packet_receive_time = self.time;
572                }
573                ClientState::SendingConnectionResponse => {
574                    debug!("client received connection keep alive packet from server");
575                    self.last_packet_receive_time = self.time;
576                    self.client_index = client_index as usize;
577                    self.max_clients = max_clients as usize;
578                    self.set_state(ClientState::Connected);
579                    info!("client connected to server");
580                }
581                _ => {}
582            },
583
584            Packet::Payload(payload) => {
585                if self.state == ClientState::Connected {
586                    debug!("client received connection payload packet from server");
587                    if self.packet_receive_queue.len() < PACKET_QUEUE_SIZE {
588                        self.packet_receive_queue.push_back((payload, sequence));
589                    }
590                    self.last_packet_receive_time = self.time;
591                }
592            }
593
594            Packet::Disconnect => {
595                if self.state == ClientState::Connected {
596                    debug!("client received disconnect packet from server");
597                    self.should_disconnect = Some(ClientState::Disconnected);
598                    self.last_packet_receive_time = self.time;
599                }
600            }
601
602            Packet::Request { .. } | Packet::Response { .. } => unreachable!(),
603        }
604    }
605}
606
607impl Drop for Client {
608    fn drop(&mut self) {
609        self.disconnect();
610    }
611}
612
613#[cfg(test)]
614mod tests {
615    use super::*;
616
617    #[test]
618    fn state_predicates() {
619        let error_states = [
620            ClientState::ConnectTokenExpired,
621            ClientState::InvalidConnectToken,
622            ClientState::ConnectionTimedOut,
623            ClientState::ConnectionResponseTimedOut,
624            ClientState::ConnectionRequestTimedOut,
625            ClientState::ConnectionDenied,
626        ];
627        for state in error_states {
628            assert!(state.is_error());
629            assert!(state.is_disconnected());
630            assert!(!state.is_connecting());
631        }
632
633        assert!(!ClientState::Disconnected.is_error());
634        assert!(ClientState::Disconnected.is_disconnected());
635        assert!(!ClientState::Disconnected.is_connecting());
636
637        for state in [ClientState::SendingConnectionRequest, ClientState::SendingConnectionResponse]
638        {
639            assert!(state.is_connecting());
640            assert!(!state.is_error());
641            assert!(!state.is_disconnected());
642        }
643
644        assert!(!ClientState::Connected.is_error());
645        assert!(!ClientState::Connected.is_disconnected());
646        assert!(!ClientState::Connected.is_connecting());
647    }
648
649    #[test]
650    fn invalid_connect_token() {
651        let mut client = Client::new("127.0.0.1:0".parse().unwrap(), 0.0).unwrap();
652        let connect_token = [0u8; CONNECT_TOKEN_BYTES];
653        assert!(client.connect(&connect_token).is_err());
654        assert_eq!(client.state(), ClientState::InvalidConnectToken);
655    }
656}
netcode/client.rs

netcode/
client.rs
