banished_net/
native.rs

1//! The native connection: a background thread holds the WebSocket and passes messages through
2//! channels, so the game never waits on the network.
3
4use std::sync::mpsc::{channel, Receiver, Sender, TryRecvError};
5use std::time::Duration;
6
7use crate::{decode, encode, FromRelay, State, ToRelay};
8
9/// A connection to a relay (`ws://host:port`).
10pub struct Connection {
11    out: Sender<ToRelay>,
12    inbox: Receiver<Result<FromRelay, String>>,
13    state: State,
14}
15
16impl Connection {
17    /// Starts connecting; [`Connection::poll`] reports what arrives.
18    pub fn connect(url: &str) -> Connection {
19        let (out, outgoing) = channel::<ToRelay>();
20        let (incoming, inbox) = channel();
21        let url = url.to_string();
22        std::thread::spawn(move || {
23            if let Err(e) = run(&url, outgoing, &incoming) {
24                let _ = incoming.send(Err(e));
25            }
26        });
27        Connection {
28            out,
29            inbox,
30            state: State::Connecting,
31        }
32    }
33
34    /// Sends a message to the relay (dropped once the connection is closed).
35    pub fn send(&mut self, message: ToRelay) {
36        let _ = self.out.send(message);
37    }
38
39    /// What arrived since the last call.
40    pub fn poll(&mut self) -> Vec<FromRelay> {
41        let mut out = Vec::new();
42        loop {
43            match self.inbox.try_recv() {
44                Ok(Ok(m)) => {
45                    self.state = State::Open;
46                    out.push(m);
47                }
48                Ok(Err(e)) => self.state = State::Closed(e),
49                Err(TryRecvError::Empty) => break,
50                Err(TryRecvError::Disconnected) => {
51                    if !matches!(self.state, State::Closed(_)) {
52                        self.state = State::Closed("disconnected".into());
53                    }
54                    break;
55                }
56            }
57        }
58        out
59    }
60
61    /// The state of the connection, as of the last [`poll`](Self::poll).
62    pub fn state(&self) -> &State {
63        &self.state
64    }
65}
66
67fn run(
68    url: &str,
69    outgoing: Receiver<ToRelay>,
70    incoming: &Sender<Result<FromRelay, String>>,
71) -> Result<(), String> {
72    use tungstenite::{stream::MaybeTlsStream, Message};
73    let (mut socket, _) = tungstenite::connect(url).map_err(|e| format!("{url}: {e}"))?;
74    if let MaybeTlsStream::Plain(s) = socket.get_ref() {
75        s.set_read_timeout(Some(Duration::from_millis(2)))
76            .map_err(|e| e.to_string())?;
77        let _ = s.set_nodelay(true);
78    }
79    loop {
80        // Send everything queued, then read what is there.
81        loop {
82            match outgoing.try_recv() {
83                Ok(m) => socket
84                    .send(Message::Binary(encode(&m).into()))
85                    .map_err(|e| e.to_string())?,
86                Err(TryRecvError::Empty) => break,
87                Err(TryRecvError::Disconnected) => {
88                    let _ = socket.close(None);
89                    return Ok(());
90                }
91            }
92        }
93        match socket.read() {
94            Ok(Message::Binary(bytes)) => {
95                let m = decode::<FromRelay>(&bytes)?;
96                if incoming.send(Ok(m)).is_err() {
97                    return Ok(());
98                }
99            }
100            Ok(Message::Close(_)) => return Err("the relay closed the connection".into()),
101            Ok(_) => {}
102            Err(tungstenite::Error::Io(e))
103                if matches!(
104                    e.kind(),
105                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
106                ) => {}
107            Err(e) => return Err(e.to_string()),
108        }
109    }
110}
banished_net/native.rs

banished_net/
native.rs
