Add documentation, add PeerConfig
This commit is contained in:
parent
b9e40ceede
commit
97230a23be
@ -9,11 +9,12 @@ use std::{
|
|||||||
use thiserror::Error;
|
use thiserror::Error;
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
PeerMessage,
|
PeerConfig, PeerMessage,
|
||||||
listener::{DATAGRAM_SIZE, Datagram, ListenerError},
|
listener::{DATAGRAM_SIZE, Datagram, ListenerError},
|
||||||
package::Package,
|
package::Package,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
/// Represents an error during sending a package for any number of reason
|
||||||
#[derive(Debug, Error)]
|
#[derive(Debug, Error)]
|
||||||
pub enum SendError {
|
pub enum SendError {
|
||||||
#[error("Serialization Error: {0}")]
|
#[error("Serialization Error: {0}")]
|
||||||
@ -26,6 +27,7 @@ pub enum SendError {
|
|||||||
SendError(std::io::Error),
|
SendError(std::io::Error),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Represents a general error which can occur specifically to a connection
|
||||||
#[derive(Debug, Error)]
|
#[derive(Debug, Error)]
|
||||||
pub enum ConnectionError {
|
pub enum ConnectionError {
|
||||||
#[error(transparent)]
|
#[error(transparent)]
|
||||||
@ -36,21 +38,27 @@ pub enum ConnectionError {
|
|||||||
TimedOut,
|
TimedOut,
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct ConnectionManager {
|
/// Manages connections for the Peer
|
||||||
|
pub(crate) struct ConnectionManager {
|
||||||
|
/// Wraps UdpSocket with some helper methods
|
||||||
pub udp: UdpWrapper,
|
pub udp: UdpWrapper,
|
||||||
connections: HashMap<SocketAddr, Connection>,
|
connections: HashMap<SocketAddr, Connection>,
|
||||||
closing_since: Option<Instant>,
|
closing_since: Option<Instant>,
|
||||||
|
config: PeerConfig,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ConnectionManager {
|
impl ConnectionManager {
|
||||||
pub fn new(socket: Arc<UdpSocket>) -> ConnectionManager {
|
/// Creates a new ConnectionManager with the given socket and config
|
||||||
|
pub fn new(socket: Arc<UdpSocket>, config: PeerConfig) -> ConnectionManager {
|
||||||
ConnectionManager {
|
ConnectionManager {
|
||||||
udp: UdpWrapper::new(socket),
|
udp: UdpWrapper::new(socket),
|
||||||
connections: HashMap::new(),
|
connections: HashMap::new(),
|
||||||
closing_since: None,
|
closing_since: None,
|
||||||
|
config,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Start closing all outbound connections.
|
||||||
pub fn close(&mut self) {
|
pub fn close(&mut self) {
|
||||||
self.closing_since = Some(Instant::now());
|
self.closing_since = Some(Instant::now());
|
||||||
for (addr, _) in self.connections.clone() {
|
for (addr, _) in self.connections.clone() {
|
||||||
@ -58,11 +66,15 @@ impl ConnectionManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Attempt to connect to a remote peer
|
||||||
pub fn connect_to(&mut self, addr: SocketAddr) {
|
pub fn connect_to(&mut self, addr: SocketAddr) {
|
||||||
self.connections
|
self.connections
|
||||||
.insert(addr, Connection::from(addr, ConnectionState::Connecting));
|
.insert(addr, Connection::from(addr, ConnectionState::Connecting));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Send possible "pings". This also does some other busywork that is
|
||||||
|
/// necessary to be run as often as possible, such as closing the udp socket
|
||||||
|
/// itself, and "hello" or "close" messages.
|
||||||
pub fn send_pings(&mut self) {
|
pub fn send_pings(&mut self) {
|
||||||
let now = Instant::now();
|
let now = Instant::now();
|
||||||
|
|
||||||
@ -71,7 +83,7 @@ impl ConnectionManager {
|
|||||||
{
|
{
|
||||||
if self.connections.len() == 0 {
|
if self.connections.len() == 0 {
|
||||||
self.udp.close();
|
self.udp.close();
|
||||||
} else if now - closing_since > Duration::from_millis(1000) {
|
} else if now - closing_since > self.config.disconnect_timeout * 2 {
|
||||||
self.udp.close();
|
self.udp.close();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -79,7 +91,7 @@ impl ConnectionManager {
|
|||||||
if let Some(_) = &self.udp.socket {
|
if let Some(_) = &self.udp.socket {
|
||||||
for (addr, conn) in &mut self.connections {
|
for (addr, conn) in &mut self.connections {
|
||||||
let duration = now - conn.last_sent_ping;
|
let duration = now - conn.last_sent_ping;
|
||||||
if duration > Duration::from_millis(100) {
|
if duration > self.config.ping_interval {
|
||||||
match conn.state {
|
match conn.state {
|
||||||
ConnectionState::ReceivingConnection | ConnectionState::Connecting => {
|
ConnectionState::ReceivingConnection | ConnectionState::Connecting => {
|
||||||
match self.udp.send_to(*addr, Package::Hello) {
|
match self.udp.send_to(*addr, Package::Hello) {
|
||||||
@ -124,6 +136,7 @@ impl ConnectionManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Close the connection to another peer at the given address
|
||||||
pub fn close_connection(&mut self, addr: &SocketAddr) {
|
pub fn close_connection(&mut self, addr: &SocketAddr) {
|
||||||
if let Some(conn) = self.connections.get_mut(addr) {
|
if let Some(conn) = self.connections.get_mut(addr) {
|
||||||
if conn.state != ConnectionState::Closing {
|
if conn.state != ConnectionState::Closing {
|
||||||
@ -133,6 +146,7 @@ impl ConnectionManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Close the connection to another peer, but with a specified error.
|
||||||
pub fn error_connection(&mut self, addr: &SocketAddr, error: ConnectionError) {
|
pub fn error_connection(&mut self, addr: &SocketAddr, error: ConnectionError) {
|
||||||
if let Some(conn) = self.connections.get_mut(addr) {
|
if let Some(conn) = self.connections.get_mut(addr) {
|
||||||
if conn.state != ConnectionState::Error {
|
if conn.state != ConnectionState::Error {
|
||||||
@ -143,6 +157,7 @@ impl ConnectionManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Handle Package from a remote peer
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn handle_package(&mut self, package: Package, addr: &SocketAddr) -> Vec<PeerMessage> {
|
pub fn handle_package(&mut self, package: Package, addr: &SocketAddr) -> Vec<PeerMessage> {
|
||||||
let mut messages = Vec::new();
|
let mut messages = Vec::new();
|
||||||
@ -220,15 +235,18 @@ impl ConnectionManager {
|
|||||||
messages
|
messages
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Close any connections that have not sent a ping for at least "timeout"s
|
||||||
|
/// duration.
|
||||||
pub fn close_silent_connections(&mut self) {
|
pub fn close_silent_connections(&mut self) {
|
||||||
let now = Instant::now();
|
let now = Instant::now();
|
||||||
for (addr, conn) in self.connections.clone() {
|
for (addr, conn) in self.connections.clone() {
|
||||||
if (now - conn.last_recv_ping) > Duration::from_millis(2000) {
|
if (now - conn.last_recv_ping) > self.config.timeout {
|
||||||
self.error_connection(&addr, ConnectionError::TimedOut);
|
self.error_connection(&addr, ConnectionError::TimedOut);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Remove any connections that have closed or errored.
|
||||||
pub fn purge_old_connections(&mut self) -> Vec<PeerMessage> {
|
pub fn purge_old_connections(&mut self) -> Vec<PeerMessage> {
|
||||||
let mut messages = Vec::new();
|
let mut messages = Vec::new();
|
||||||
|
|
||||||
@ -237,7 +255,7 @@ impl ConnectionManager {
|
|||||||
self.connections.retain(|_, conn| {
|
self.connections.retain(|_, conn| {
|
||||||
let retain = match conn.state {
|
let retain = match conn.state {
|
||||||
ConnectionState::ReceivingClosing => {
|
ConnectionState::ReceivingClosing => {
|
||||||
if (now - conn.last_recv_close) > Duration::from_millis(500) {
|
if (now - conn.last_recv_close) > self.config.disconnect_timeout {
|
||||||
messages.push(PeerMessage::Disconnected(conn.clone(), None));
|
messages.push(PeerMessage::Disconnected(conn.clone(), None));
|
||||||
false
|
false
|
||||||
} else {
|
} else {
|
||||||
@ -257,7 +275,8 @@ impl ConnectionManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct UdpWrapper {
|
/// Wrap UdpSocket with some helper methods
|
||||||
|
pub(crate) struct UdpWrapper {
|
||||||
socket: Option<Arc<UdpSocket>>,
|
socket: Option<Arc<UdpSocket>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -268,14 +287,17 @@ impl UdpWrapper {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Close the method, that is to say, drop it.
|
||||||
pub fn close(&mut self) {
|
pub fn close(&mut self) {
|
||||||
self.socket = None;
|
self.socket = None;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Is the socket closed
|
||||||
pub fn is_closed(&self) -> bool {
|
pub fn is_closed(&self) -> bool {
|
||||||
self.socket.is_none()
|
self.socket.is_none()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Send a Package to a remote peer at addr
|
||||||
fn send_to(&self, addr: SocketAddr, package: Package) -> Result<(), SendError> {
|
fn send_to(&self, addr: SocketAddr, package: Package) -> Result<(), SendError> {
|
||||||
if let Some(socket) = &self.socket {
|
if let Some(socket) = &self.socket {
|
||||||
let mut buf = Vec::new();
|
let mut buf = Vec::new();
|
||||||
@ -295,8 +317,12 @@ impl UdpWrapper {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Represents a single connection from the peer to another peer
|
||||||
pub struct Connection {
|
pub struct Connection {
|
||||||
|
/// Address of the connection
|
||||||
pub address: SocketAddr,
|
pub address: SocketAddr,
|
||||||
|
/// "ping", as in the time between the last sent ping and the last received
|
||||||
|
/// pong after it.
|
||||||
pub ping: Duration,
|
pub ping: Duration,
|
||||||
|
|
||||||
last_sent_ping: Instant,
|
last_sent_ping: Instant,
|
||||||
@ -304,6 +330,7 @@ pub struct Connection {
|
|||||||
last_recv_close: Instant,
|
last_recv_close: Instant,
|
||||||
closing_since: Instant,
|
closing_since: Instant,
|
||||||
|
|
||||||
|
/// What state is the connection currently in?
|
||||||
pub state: ConnectionState,
|
pub state: ConnectionState,
|
||||||
error: Option<ConnectionError>,
|
error: Option<ConnectionError>,
|
||||||
}
|
}
|
||||||
@ -338,15 +365,24 @@ impl Connection {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// State of the connection
|
||||||
#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Copy)]
|
#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Copy)]
|
||||||
pub enum ConnectionState {
|
pub enum ConnectionState {
|
||||||
|
/// Connection is currently attempting to connect to another peer
|
||||||
Connecting,
|
Connecting,
|
||||||
|
/// Connection is attepting to connect to another peer, and has received at
|
||||||
|
/// least one "hello" in response.
|
||||||
ConnectingNearlyReady,
|
ConnectingNearlyReady,
|
||||||
|
/// Connection is receiving a connection from another peer
|
||||||
ReceivingConnection,
|
ReceivingConnection,
|
||||||
|
/// Connection has been properly established
|
||||||
Connected,
|
Connected,
|
||||||
|
|
||||||
|
/// Connection is being closed
|
||||||
Closing,
|
Closing,
|
||||||
|
/// Connection is being closed by the other peer
|
||||||
ReceivingClosing,
|
ReceivingClosing,
|
||||||
|
|
||||||
|
/// Connection has ended up in an erronous state
|
||||||
Error,
|
Error,
|
||||||
}
|
}
|
||||||
|
|||||||
60
src/lib.rs
60
src/lib.rs
@ -7,6 +7,7 @@ use std::{
|
|||||||
mpsc::{Receiver, channel},
|
mpsc::{Receiver, channel},
|
||||||
},
|
},
|
||||||
thread::spawn,
|
thread::spawn,
|
||||||
|
time::Duration,
|
||||||
};
|
};
|
||||||
|
|
||||||
use thiserror::*;
|
use thiserror::*;
|
||||||
@ -20,6 +21,7 @@ pub mod connections;
|
|||||||
pub(crate) mod listener;
|
pub(crate) mod listener;
|
||||||
pub(crate) mod package;
|
pub(crate) mod package;
|
||||||
|
|
||||||
|
/// Error for the Peer
|
||||||
#[derive(Error, Debug)]
|
#[derive(Error, Debug)]
|
||||||
pub enum PeerError {
|
pub enum PeerError {
|
||||||
#[error("Error binding to port: {0}")]
|
#[error("Error binding to port: {0}")]
|
||||||
@ -28,12 +30,61 @@ pub enum PeerError {
|
|||||||
ListenerError(std::io::Error),
|
ListenerError(std::io::Error),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Possible messages from occurring events from the peer
|
||||||
pub enum PeerMessage {
|
pub enum PeerMessage {
|
||||||
|
/// A new connection has been connected
|
||||||
NewConnection(Connection),
|
NewConnection(Connection),
|
||||||
|
/// An existing connection has disconnected, with an optional error
|
||||||
Disconnected(Connection, Option<ConnectionError>),
|
Disconnected(Connection, Option<ConnectionError>),
|
||||||
|
/// The Peer has been closed
|
||||||
Closed,
|
Closed,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Optional configuration available for peers, mainly configuration of
|
||||||
|
/// different intervals
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub struct PeerConfig {
|
||||||
|
ping_interval: Duration,
|
||||||
|
timeout: Duration,
|
||||||
|
disconnect_timeout: Duration,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for PeerConfig {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
ping_interval: Duration::from_millis(100),
|
||||||
|
timeout: Duration::from_millis(2000),
|
||||||
|
disconnect_timeout: Duration::from_millis(500),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl PeerConfig {
|
||||||
|
/// Set the desired interval between "ping"s
|
||||||
|
pub fn with_ping_interval(self, interval: Duration) -> PeerConfig {
|
||||||
|
PeerConfig {
|
||||||
|
ping_interval: interval,
|
||||||
|
..self
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Set the length of timeout; that is to say the duration of time which is
|
||||||
|
/// acceptable to occur between pings before a Timeout-error occurs
|
||||||
|
pub fn with_timeout(self, timeout: Duration) -> PeerConfig {
|
||||||
|
PeerConfig { timeout, ..self }
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Set the length of time that is awaited since the last "Closed" message
|
||||||
|
/// during disconnection before disconnection actually occurs. Should be
|
||||||
|
/// significantly higher than ping interval.
|
||||||
|
pub fn with_disconnect(self, timeout: Duration) -> PeerConfig {
|
||||||
|
PeerConfig {
|
||||||
|
disconnect_timeout: timeout,
|
||||||
|
..self
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub struct Peer {
|
pub struct Peer {
|
||||||
connection_mgr: ConnectionManager,
|
connection_mgr: ConnectionManager,
|
||||||
closed: Arc<AtomicBool>,
|
closed: Arc<AtomicBool>,
|
||||||
@ -44,7 +95,7 @@ pub struct Peer {
|
|||||||
impl Peer {
|
impl Peer {
|
||||||
/// Bind given port to listen to incoming messages. Creates a new Peer that
|
/// Bind given port to listen to incoming messages. Creates a new Peer that
|
||||||
/// can establish new connections.
|
/// can establish new connections.
|
||||||
pub fn listen(port: Option<u16>) -> Result<Peer, PeerError> {
|
pub fn listen(port: Option<u16>, config: PeerConfig) -> Result<Peer, PeerError> {
|
||||||
let socket = Arc::new(
|
let socket = Arc::new(
|
||||||
UdpSocket::bind(SocketAddr::from(([0, 0, 0, 0], port.unwrap_or(0))))
|
UdpSocket::bind(SocketAddr::from(([0, 0, 0, 0], port.unwrap_or(0))))
|
||||||
.map_err(|e| PeerError::BindError(e))?,
|
.map_err(|e| PeerError::BindError(e))?,
|
||||||
@ -66,17 +117,21 @@ impl Peer {
|
|||||||
});
|
});
|
||||||
|
|
||||||
Ok(Peer {
|
Ok(Peer {
|
||||||
connection_mgr: ConnectionManager::new(socket),
|
connection_mgr: ConnectionManager::new(socket, config.clone()),
|
||||||
closed,
|
closed,
|
||||||
receiver,
|
receiver,
|
||||||
messages: VecDeque::new(),
|
messages: VecDeque::new(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Connect the peer to the given address.
|
||||||
pub fn connect_to(&mut self, addr: SocketAddr) {
|
pub fn connect_to(&mut self, addr: SocketAddr) {
|
||||||
self.connection_mgr.connect_to(addr);
|
self.connection_mgr.connect_to(addr);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Send pings and poll for events from the listener thread. Returns
|
||||||
|
/// possible events, such as connections and messages. Should be called as
|
||||||
|
/// often as possible.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn poll(&mut self) -> Result<Option<PeerMessage>, PeerError> {
|
pub fn poll(&mut self) -> Result<Option<PeerMessage>, PeerError> {
|
||||||
if self.connection_mgr.udp.is_closed() {
|
if self.connection_mgr.udp.is_closed() {
|
||||||
@ -125,6 +180,7 @@ impl Peer {
|
|||||||
Ok(self.messages.pop_front())
|
Ok(self.messages.pop_front())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Close the Peer, disconnecting all connections.
|
||||||
pub fn close(&mut self) {
|
pub fn close(&mut self) {
|
||||||
self.connection_mgr.close();
|
self.connection_mgr.close();
|
||||||
}
|
}
|
||||||
|
|||||||
@ -11,6 +11,7 @@ use crate::package::Package;
|
|||||||
pub const DATAGRAM_SIZE: usize = 102400;
|
pub const DATAGRAM_SIZE: usize = 102400;
|
||||||
pub type Datagram = [u8; DATAGRAM_SIZE];
|
pub type Datagram = [u8; DATAGRAM_SIZE];
|
||||||
|
|
||||||
|
/// Listener thread for Peer's UdpSocket
|
||||||
pub struct Listener {
|
pub struct Listener {
|
||||||
socket: Arc<UdpSocket>,
|
socket: Arc<UdpSocket>,
|
||||||
sender: Sender<ListenerMessage>,
|
sender: Sender<ListenerMessage>,
|
||||||
@ -21,6 +22,7 @@ impl Listener {
|
|||||||
Listener { socket, sender }
|
Listener { socket, sender }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Read from the socket, parse packages and emit messages.
|
||||||
pub fn poll(&mut self) {
|
pub fn poll(&mut self) {
|
||||||
let mut buffer: Datagram = [0; _];
|
let mut buffer: Datagram = [0; _];
|
||||||
match self.socket.recv_from(&mut buffer) {
|
match self.socket.recv_from(&mut buffer) {
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user