use crate::errors::*;
use crate::net::Connection;
use crate::protocol::{Request, Response};
use crate::util;
use backon::BlockingRetryable;
use byteorder::{BigEndian, ByteOrder};
use std::io::{self, BufReader, BufWriter, Read};
pub struct ServerConnection {
reader: BufReader<Box<dyn Connection>>,
writer: BufWriter<Box<dyn Connection>>,
}
impl ServerConnection {
pub fn new(conn: Box<dyn Connection>) -> io::Result<ServerConnection> {
let write_conn = conn.try_clone()?;
Ok(ServerConnection {
reader: BufReader::new(conn),
writer: BufWriter::new(write_conn),
})
}
pub fn request(&mut self, request: Request) -> Result<Response> {
trace!("ServerConnection::request");
util::write_length_prefixed_bincode(&mut self.writer, request)?;
trace!("ServerConnection::request: sent request");
self.read_one_response()
}
pub fn read_one_response(&mut self) -> Result<Response> {
trace!("ServerConnection::read_one_response");
let mut bytes = [0; 4];
self.reader
.read_exact(&mut bytes)
.context("Failed to read response header")?;
let len = BigEndian::read_u32(&bytes);
trace!("Should read {} more bytes", len);
let mut data = vec![0; len as usize];
self.reader.read_exact(&mut data)?;
trace!("Done reading");
Ok(bincode::deserialize(&data)?)
}
}
pub fn connect_to_server(addr: &crate::net::SocketAddr) -> io::Result<ServerConnection> {
trace!("connect_to_server({addr})");
let conn = crate::net::connect(addr)?;
ServerConnection::new(conn)
}
pub fn connect_with_retry(addr: &crate::net::SocketAddr) -> io::Result<ServerConnection> {
trace!("connect_with_retry({addr})");
let backoff = backon::ConstantBuilder::default()
.with_delay(std::time::Duration::from_millis(500))
.with_max_times(10);
match (|| connect_to_server(addr)).retry(backoff).call() {
Ok(conn) => Ok(conn),
Err(e) => Err(io::Error::new(
io::ErrorKind::TimedOut,
format!("Connection to server timed out: {:?}", e),
)),
}
}