mod crypto;
mod packet;
mod wire;
use std::collections::BTreeMap;
use std::fmt;
use std::net::{IpAddr, TcpListener, UdpSocket};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, mpsc};
pub use packet::{Arg, Packet};
use crate::error::{Error, Result};
use wire::{Event, Frame, Link};
pub const DEFAULT_PORT: u16 = 7777;
const MAX_NAME: usize = 32;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default)]
pub struct PeerId(u32);
impl PeerId {
pub const SERVER: PeerId = PeerId(0);
pub fn number(self) -> u32 {
self.0
}
pub fn from_number(number: u32) -> Self {
Self(number)
}
pub fn is_server(self) -> bool {
self == Self::SERVER
}
}
impl fmt::Display for PeerId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}", self.0)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Reason {
Left,
Closed,
Kicked(String),
ServerFull,
WrongGame(String),
WrongVersion(String),
TimedOut,
Failed(String),
Lost(String),
}
impl fmt::Display for Reason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&crate::lang::Lang::new().describe_reason(self))
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum NetEvent {
Connected(PeerId),
Message(PeerId, Packet),
Disconnected(PeerId, Reason),
}
enum State {
Offline,
Hosting { port: u16, stop: Arc<AtomicBool> },
Joining,
Joined,
}
struct Peer {
name: String,
link: Link,
ping: f32,
}
pub struct Net {
state: State,
events_rx: Option<Receiver<Event>>,
peers: BTreeMap<PeerId, Peer>,
pub(crate) events: Vec<NetEvent>,
pending: Vec<NetEvent>,
game: String,
version: String,
name: String,
max_players: usize,
id: PeerId,
warned_offline: bool,
}
impl fmt::Debug for Net {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let state = match self.state {
State::Offline => "offline",
State::Hosting { .. } => "hosting",
State::Joining => "joining",
State::Joined => "joined",
};
f.debug_struct("Net").field("state", &state).field("players", &self.peers.len()).finish_non_exhaustive()
}
}
impl Net {
pub fn new(game: impl Into<String>) -> Self {
Self {
state: State::Offline,
events_rx: None,
peers: BTreeMap::new(),
events: Vec::new(),
pending: Vec::new(),
game: game.into(),
version: String::new(),
name: "Player".into(),
max_players: 32,
id: PeerId::SERVER,
warned_offline: false,
}
}
pub fn set_name(&mut self, name: impl Into<String>) {
self.name = clean_name(&name.into());
}
pub fn set_version(&mut self, version: impl Into<String>) {
self.version = version.into();
}
pub fn set_max_players(&mut self, max: usize) {
self.max_players = max;
}
pub fn host(&mut self, port: u16) -> Result<()> {
self.close();
let listener = TcpListener::bind(("0.0.0.0", port))
.map_err(|e| Error::Net(format!("could not host on port {port}: {e}")))?;
let port = listener.local_addr().map(|a| a.port()).unwrap_or(port);
listener.set_nonblocking(true)?;
let stop = Arc::new(AtomicBool::new(false));
let tx = self.new_channel();
let flag = stop.clone();
std::thread::Builder::new()
.name("audio-code net server".into())
.spawn(move || wire::accept_loop(listener, flag, tx))?;
self.state = State::Hosting { port, stop };
self.id = PeerId::SERVER;
Ok(())
}
pub fn join(&mut self, address: &str) {
self.close();
let tx = self.new_channel();
let hello = Frame::Hello {
protocol: wire::PROTOCOL,
game: self.game.clone(),
version: self.version.clone(),
name: self.name.clone(),
};
let address = address.to_string();
let spawned = std::thread::Builder::new()
.name("audio-code net client".into())
.spawn(move || wire::connect(address, hello, tx));
if let Err(e) = spawned {
self.pending.push(NetEvent::Disconnected(PeerId::SERVER, Reason::Failed(e.to_string())));
return;
}
self.state = State::Joining;
}
pub fn close(&mut self) {
let bye = if self.is_host() { Reason::Closed } else { Reason::Left };
for peer in std::mem::take(&mut self.peers).into_values() {
peer.link.close(Some(bye.clone()));
}
if let State::Hosting { stop, .. } = &self.state {
stop.store(true, Ordering::Relaxed);
}
self.state = State::Offline;
self.events_rx = None;
self.id = PeerId::SERVER;
self.warned_offline = false;
}
pub fn kick(&mut self, peer: PeerId, reason: &str) {
if !self.is_host() {
return;
}
if let Some(p) = self.peers.remove(&peer) {
p.link.close(Some(Reason::Kicked(reason.to_string())));
self.pending.push(NetEvent::Disconnected(peer, Reason::Kicked(reason.to_string())));
}
}
pub fn send(&mut self, peer: PeerId, packet: impl Into<Packet>) {
let bytes = encode(packet.into());
match self.peers.get(&peer) {
Some(p) => p.link.send(bytes),
None => self.warn_offline(),
}
}
pub fn send_all(&mut self, packet: impl Into<Packet>) {
self.send_filtered(packet.into(), |_| true);
}
pub fn send_all_except(&mut self, except: PeerId, packet: impl Into<Packet>) {
self.send_filtered(packet.into(), |p| p != except);
}
fn send_filtered(&mut self, packet: Packet, keep: impl Fn(PeerId) -> bool) {
if self.peers.is_empty() {
if !self.is_host() {
self.warn_offline();
}
return;
}
let bytes = encode(packet);
for (_, p) in self.peers.iter().filter(|(id, _)| keep(**id)) {
p.link.send(bytes.clone());
}
}
fn warn_offline(&mut self) {
if !self.warned_offline && !matches!(self.state, State::Joined | State::Hosting { .. }) {
self.warned_offline = true;
eprintln!("audio-code: a packet was sent while not connected; it was dropped");
}
}
pub fn events(&self) -> &[NetEvent] {
&self.events
}
pub fn joined(&self) -> Vec<PeerId> {
self.events
.iter()
.filter_map(|e| match e {
NetEvent::Connected(p) => Some(*p),
_ => None,
})
.collect()
}
pub fn left(&self) -> Vec<(PeerId, Reason)> {
self.events
.iter()
.filter_map(|e| match e {
NetEvent::Disconnected(p, r) => Some((*p, r.clone())),
_ => None,
})
.collect()
}
pub fn messages(&self) -> Vec<(PeerId, Packet)> {
self.events
.iter()
.filter_map(|e| match e {
NetEvent::Message(p, packet) => Some((*p, packet.clone())),
_ => None,
})
.collect()
}
pub fn received(&self, kind: &str) -> Vec<(PeerId, Packet)> {
self.events
.iter()
.filter_map(|e| match e {
NetEvent::Message(p, packet) if packet.is(kind) => Some((*p, packet.clone())),
_ => None,
})
.collect()
}
pub fn players(&self) -> Vec<PeerId> {
self.peers.keys().copied().collect()
}
pub fn player_count(&self) -> usize {
self.peers.len()
}
pub fn is_connected_to(&self, peer: PeerId) -> bool {
self.peers.contains_key(&peer)
}
pub fn name(&self, peer: PeerId) -> &str {
if peer == self.id && !matches!(self.state, State::Offline | State::Joining) {
return &self.name;
}
self.peers.get(&peer).map_or("", |p| p.name.as_str())
}
pub fn ping(&self, peer: PeerId) -> f32 {
self.peers.get(&peer).map_or(0.0, |p| p.ping)
}
pub fn id(&self) -> PeerId {
self.id
}
pub fn is_host(&self) -> bool {
matches!(self.state, State::Hosting { .. })
}
pub fn is_client(&self) -> bool {
matches!(self.state, State::Joined)
}
pub fn is_joining(&self) -> bool {
matches!(self.state, State::Joining)
}
pub fn is_online(&self) -> bool {
self.is_host() || self.is_client()
}
pub fn port(&self) -> Option<u16> {
match self.state {
State::Hosting { port, .. } => Some(port),
_ => None,
}
}
pub fn local_ip() -> Option<IpAddr> {
let socket = UdpSocket::bind("0.0.0.0:0").ok()?;
socket.connect("8.8.8.8:80").ok()?;
socket.local_addr().ok().map(|a| a.ip()).filter(|ip| !ip.is_unspecified())
}
fn new_channel(&mut self) -> Sender<Event> {
let (tx, rx) = mpsc::channel();
self.events_rx = Some(rx);
tx
}
pub(crate) fn update(&mut self) {
self.events = std::mem::take(&mut self.pending);
let Some(rx) = &self.events_rx else {
return;
};
let reports: Vec<Event> = rx.try_iter().collect();
for report in reports {
self.handle(report);
}
}
fn handle(&mut self, report: Event) {
match report {
Event::Hello { link, protocol, game, version, name } => {
if !self.is_host() {
link.close(None);
return;
}
let refusal = if game != self.game {
Some(Reason::WrongGame(self.game.clone()))
} else if version != self.version || protocol != wire::PROTOCOL {
Some(Reason::WrongVersion(self.version.clone()))
} else if self.peers.len() >= self.max_players {
Some(Reason::ServerFull)
} else {
None
};
if let Some(reason) = refusal {
link.close(Some(reason));
return;
}
let peer = PeerId(link.id);
link.send_frame(&Frame::Welcome { id: link.id, name: self.name.clone() });
self.peers.insert(peer, Peer { name: clean_name(&name), link, ping: 0.0 });
self.events.push(NetEvent::Connected(peer));
}
Event::Welcome { link, id, name } => {
self.state = State::Joined;
self.id = PeerId(id);
self.peers.insert(PeerId::SERVER, Peer { name: clean_name(&name), link, ping: 0.0 });
self.events.push(NetEvent::Connected(PeerId::SERVER));
}
Event::Data { peer, packet } => {
let peer = PeerId(peer);
if self.peers.contains_key(&peer) {
self.events.push(NetEvent::Message(peer, packet));
}
}
Event::Ping { peer, seconds } => {
if let Some(p) = self.peers.get_mut(&PeerId(peer)) {
p.ping = seconds;
}
}
Event::Closed { peer, reason } => {
let peer = PeerId(peer);
let known = self.peers.remove(&peer).is_some();
let joining = self.is_joining() && peer == PeerId::SERVER;
if known || joining {
self.events.push(NetEvent::Disconnected(peer, reason));
}
if !self.is_host() && peer == PeerId::SERVER {
self.state = State::Offline;
self.events_rx = None;
self.id = PeerId::SERVER;
}
}
}
}
}
impl Drop for Net {
fn drop(&mut self) {
self.close();
}
}
fn encode(packet: Packet) -> Arc<[u8]> {
let bytes = Frame::Data(packet).encode();
if bytes.len() > wire::MAX_FRAME {
eprintln!("audio-code: a packet of {} bytes is too large to send (at most 1 MB); it was dropped", bytes.len());
return Arc::from(&[][..]);
}
bytes.into()
}
fn clean_name(name: &str) -> String {
name.chars().filter(|c| !c.is_control()).take(MAX_NAME).collect::<String>().trim().to_string()
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::{Duration, Instant};
fn pump(nets: &mut [&mut Net], mut done: impl FnMut(&[&mut Net]) -> bool) -> Vec<Vec<NetEvent>> {
let mut all = vec![Vec::new(); nets.len()];
let start = Instant::now();
while start.elapsed() < Duration::from_secs(8) {
for (net, events) in nets.iter_mut().zip(&mut all) {
net.update();
events.extend(net.events.iter().cloned());
}
if done(nets) {
break;
}
std::thread::sleep(Duration::from_millis(5));
}
all
}
fn host() -> (Net, String) {
let mut server = Net::new("game");
server.set_name("Host");
server.host(0).unwrap();
let address = format!("127.0.0.1:{}", server.port().unwrap());
(server, address)
}
#[test]
fn join_chat_and_leave() {
let (mut server, address) = host();
let mut client = Net::new("game");
client.set_name("Anna");
client.join(&address);
pump(&mut [&mut server, &mut client], |n| n[0].player_count() == 1 && n[1].is_client());
let anna = server.players()[0];
assert_eq!(server.name(anna), "Anna");
assert_eq!(client.name(PeerId::SERVER), "Host");
assert_eq!(client.id(), anna);
assert_eq!(client.name(client.id()), "Anna");
client.send_all(Packet::new("chat").with("hi"));
let events = pump(&mut [&mut server, &mut client], |n| !n[0].received("chat").is_empty());
assert_eq!(events[0], [NetEvent::Message(anna, Packet::new("chat").with("hi"))]);
server.send(anna, "pong");
let events = pump(&mut [&mut server, &mut client], |n| !n[1].messages().is_empty());
assert_eq!(events[1], [NetEvent::Message(PeerId::SERVER, Packet::new("pong"))]);
client.close();
let events = pump(&mut [&mut server], |n| n[0].player_count() == 0);
assert_eq!(events[0], [NetEvent::Disconnected(anna, Reason::Left)]);
}
#[test]
fn refuses_wrong_game_version_and_full() {
let (mut server, address) = host();
server.set_version("2");
server.set_max_players(1);
let mut other = Net::new("other game");
other.join(&address);
let mut old = Net::new("game");
old.set_version("1");
old.join(&address);
let events = pump(&mut [&mut server, &mut other, &mut old], |n| !n[1].is_joining() && !n[2].is_joining());
assert_eq!(events[1], [NetEvent::Disconnected(PeerId::SERVER, Reason::WrongGame("game".into()))]);
assert_eq!(events[2], [NetEvent::Disconnected(PeerId::SERVER, Reason::WrongVersion("2".into()))]);
let mut first = Net::new("game");
first.set_version("2");
first.join(&address);
pump(&mut [&mut server, &mut first], |n| n[1].is_client());
let mut second = Net::new("game");
second.set_version("2");
second.join(&address);
let events = pump(&mut [&mut server, &mut second], |n| !n[1].is_joining());
assert_eq!(events[1], [NetEvent::Disconnected(PeerId::SERVER, Reason::ServerFull)]);
}
#[test]
fn kick_and_server_close() {
let (mut server, address) = host();
let mut a = Net::new("game");
let mut b = Net::new("game");
a.join(&address);
b.join(&address);
pump(&mut [&mut server, &mut a, &mut b], |n| n[0].player_count() == 2 && n[1].is_client() && n[2].is_client());
server.kick(a.id(), "spam");
let events = pump(&mut [&mut server, &mut a], |n| !n[1].is_online());
assert_eq!(events[1], [NetEvent::Disconnected(PeerId::SERVER, Reason::Kicked("spam".into()))]);
server.close();
let events = pump(&mut [&mut b], |n| !n[0].is_online());
assert_eq!(events[0], [NetEvent::Disconnected(PeerId::SERVER, Reason::Closed)]);
}
#[test]
fn join_failure_is_an_event() {
let port = TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
let mut client = Net::new("game");
client.join(&format!("127.0.0.1:{port}"));
let events = pump(&mut [&mut client], |n| !n[0].is_joining());
assert!(matches!(events[0].as_slice(), [NetEvent::Disconnected(PeerId::SERVER, Reason::Failed(_))]));
client.send_all("hello");
}
}