//! The client side: a connection to another device's app on its own thread,
//! reconnecting when it drops. Messages from the host reach the workspace as
//! [`ClientEvent`]s.
use std::{
io::ErrorKind,
net::{TcpStream, ToSocketAddrs},
sync::{
Arc,
atomic::{AtomicBool, Ordering},
mpsc,
},
thread,
time::Duration,
};
use tungstenite::{Message, WebSocket};
use super::{ClientMessage, PROTOCOL, ServerMessage, with_port};
const CONNECT_TIMEOUT: Duration = Duration::from_secs(6);
const POLL: Duration = Duration::from_millis(40);
#[derive(Clone, Debug, PartialEq)]
pub enum ConnectionStatus {
Connecting,
/// Connected to the host of that name.
Connected(String),
Failed(String),
/// The host refused the key; no more retries until the settings change.
Denied(String),
}
pub enum ClientEvent {
Status(ConnectionStatus),
Message(ServerMessage),
}
pub struct RemoteClient {
outbox: mpsc::Sender<String>,
stop: Arc<AtomicBool>,
pub events: smol::channel::Receiver<ClientEvent>,
}
impl RemoteClient {
/// Connects to `address` (`host` or `host:port`) and keeps reconnecting
/// until dropped.
pub fn connect(address: &str, key: String, name: String) -> Self {
let address = with_port(address);
let stop = Arc::new(AtomicBool::new(false));
let (outbox, inbox) = mpsc::channel::<String>();
let (events_tx, events) = smol::channel::unbounded();
{
let stop = stop.clone();
let _ = thread::Builder::new().name("remote-client".into()).spawn(move || {
let mut backoff = Duration::from_secs(2);
while !stop.load(Ordering::Relaxed) {
let _ = events_tx.send_blocking(ClientEvent::Status(ConnectionStatus::Connecting));
match open(&address, &key, &name) {
Ok((mut socket, host)) => {
backoff = Duration::from_secs(2);
// Whatever was asked while offline is stale now.
while inbox.try_recv().is_ok() {}
let _ = events_tx.send_blocking(ClientEvent::Status(ConnectionStatus::Connected(host)));
let reason = run(&mut socket, &inbox, &events_tx, &stop);
let _ = socket.close(None);
if stop.load(Ordering::Relaxed) {
break;
}
let _ = events_tx.send_blocking(ClientEvent::Status(ConnectionStatus::Failed(reason)));
}
Err(Failure::Denied(reason)) => {
let _ = events_tx.send_blocking(ClientEvent::Status(ConnectionStatus::Denied(reason)));
break;
}
Err(Failure::Other(reason)) => {
let _ = events_tx.send_blocking(ClientEvent::Status(ConnectionStatus::Failed(reason)));
}
}
// Wait before retrying, a little longer each time.
let mut waited = Duration::ZERO;
while waited < backoff && !stop.load(Ordering::Relaxed) {
thread::sleep(Duration::from_millis(200));
waited += Duration::from_millis(200);
}
backoff = (backoff * 2).min(Duration::from_secs(30));
}
});
}
Self { outbox, stop, events }
}
pub fn send(&self, message: &ClientMessage) {
if let Ok(text) = serde_json::to_string(message) {
let _ = self.outbox.send(text);
}
}
}
impl Drop for RemoteClient {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
}
}
enum Failure {
Denied(String),
Other(String),
}
/// Connects, says hello, and waits for the welcome.
fn open(address: &str, key: &str, name: &str) -> Result<(WebSocket<TcpStream>, String), Failure> {
let other = |err: &dyn std::fmt::Display| Failure::Other(err.to_string());
let target = address.to_socket_addrs().map_err(|err| other(&err))?.next().ok_or_else(|| other(&"no such host"))?;
let stream = TcpStream::connect_timeout(&target, CONNECT_TIMEOUT).map_err(|err| other(&err))?;
let _ = stream.set_nodelay(true);
stream.set_read_timeout(Some(Duration::from_secs(15))).map_err(|err| other(&err))?;
let (mut socket, _) = tungstenite::client(format!("ws://{address}/"), stream).map_err(|err| other(&err))?;
let hello = ClientMessage::Hello { key: key.to_string(), name: name.to_string(), protocol: PROTOCOL };
socket
.send(Message::text(serde_json::to_string(&hello).unwrap_or_default()))
.map_err(|err| other(&err))?;
loop {
match socket.read().map_err(|err| other(&err))? {
Message::Text(text) => {
return match serde_json::from_str::<ServerMessage>(text.as_str()) {
Ok(ServerMessage::Welcome { host, .. }) => {
let _ = socket.get_mut().set_read_timeout(Some(POLL));
Ok((socket, host))
}
Ok(ServerMessage::Denied { reason }) => Err(Failure::Denied(reason)),
_ => Err(other(&"the host answered something unexpected")),
};
}
Message::Close(_) => return Err(Failure::Denied("The host closed the connection.".into())),
_ => {}
}
}
}
/// Trades messages until the connection drops; returns why it did.
fn run(
socket: &mut WebSocket<TcpStream>,
inbox: &mpsc::Receiver<String>,
events: &smol::channel::Sender<ClientEvent>,
stop: &AtomicBool,
) -> String {
while !stop.load(Ordering::Relaxed) {
while let Ok(text) = inbox.try_recv() {
if let Err(err) = socket.send(Message::text(text)) {
return err.to_string();
}
}
match socket.read() {
Ok(Message::Text(text)) => match serde_json::from_str::<ServerMessage>(text.as_str()) {
Ok(message) => {
if events.send_blocking(ClientEvent::Message(message)).is_err() {
return "closed".into();
}
}
Err(err) => eprintln!("vectrum-code: unreadable message from host: {err}"),
},
Ok(Message::Close(_)) => return "The host closed the connection.".into(),
Ok(_) => {}
Err(tungstenite::Error::Io(err)) if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
Err(err) => return err.to_string(),
}
}
"closed".into()
}