use super::*;
pub const LOBBY_PORTS: [u16; 8] = [47700, 47701, 47702, 47703, 47704, 47705, 47706, 47707];
pub(super) const ANNOUNCE_MAGIC: &[u8; 5] = b"PNCH1";
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Beacon {
Here {
id: u64,
name: String,
host: String,
taken: u8,
seats: u8,
running: bool,
},
Closing { id: u64 },
}
impl Beacon {
pub fn id(&self) -> u64 {
match self {
Beacon::Here { id, .. } | Beacon::Closing { id } => *id,
}
}
pub fn has_room(&self) -> bool {
match self {
Beacon::Here { seats: 0, .. } => true,
Beacon::Here { taken, seats, .. } => taken < seats,
Beacon::Closing { .. } => false,
}
}
}
pub(super) const BEACON_OPEN: u8 = 0;
pub(super) const BEACON_CLOSING: u8 = 1;
pub(super) const BEACON_RUNNING: u8 = 2;
pub(super) const BEACON_NAME_AT: usize = 8;
pub(super) const BEACON_TAKEN_AT: usize = BEACON_NAME_AT + WIRE_NAME;
pub(super) const BEACON_SEATS_AT: usize = BEACON_TAKEN_AT + 1;
pub(super) const BEACON_ID_AT: usize = BEACON_SEATS_AT + 1;
pub(super) const BEACON_TABLE_BYTES: usize = BEACON_ID_AT + 8;
pub(super) const BEACON_HOST_AT: usize = BEACON_TABLE_BYTES;
pub(super) const BEACON_BYTES: usize = BEACON_HOST_AT + WIRE_NAME;
pub(super) const MAX_BEACON: usize = 96;
pub(super) const FAREWELL_REPEATS: usize = 3;
pub(super) const WIDE_EVERY: u32 = 4;
pub(super) const SUBNET_RECHECK_EVERY: u32 = 8;
pub(super) fn due_for_recheck(round: u32) -> bool {
round != 0 && round.is_multiple_of(SUBNET_RECHECK_EVERY)
}
pub(super) fn wide_ports(round: u32, farewell: bool) -> &'static [u16] {
match farewell || round.is_multiple_of(WIDE_EVERY) {
true => &LOBBY_PORTS,
false => &LOBBY_PORTS[..1],
}
}
pub(super) fn beacon_name(buf: &[u8], len: usize, at: usize) -> String {
buf.get(at..at + WIRE_NAME)
.filter(|_| len >= at + WIRE_NAME)
.and_then(|bytes| WireName::try_from(bytes).ok())
.map(|wire| name_from_wire(&wire))
.unwrap_or_default()
}
#[derive(Clone, Copy, Default)]
pub struct OnAir<'a> {
pub name: &'a str,
pub host: &'a str,
pub taken: u8,
pub seats: u8,
}
pub struct Discovery {
socket: UdpSocket,
}
impl Discovery {
pub fn bind() -> io::Result<Discovery> {
let mut last_err = io::Error::new(io::ErrorKind::AddrInUse, "no lobby port free");
for port in LOBBY_PORTS {
match UdpSocket::bind(("0.0.0.0", port)) {
Ok(socket) => {
socket.set_nonblocking(true)?;
return Ok(Discovery { socket });
}
Err(e) => last_err = e,
}
}
Err(last_err)
}
pub fn poll(&mut self) -> Vec<(SocketAddr, Beacon)> {
let mut out = Vec::new();
let mut buf = [0u8; MAX_BEACON];
loop {
match self.socket.recv_from(&mut buf) {
Ok((len, from)) => {
if len >= 7 && &buf[..5] == ANNOUNCE_MAGIC {
let game_port = u16::from_le_bytes([buf[5], buf[6]]);
let id = match len >= BEACON_TABLE_BYTES {
true => u64::from_le_bytes(
buf[BEACON_ID_AT..BEACON_TABLE_BYTES]
.try_into()
.expect("eight bytes"),
),
false => 0,
};
let kind = (len >= 8).then(|| buf[7]);
let beacon = match kind {
Some(BEACON_CLOSING) => Beacon::Closing { id },
kind => {
let (taken, seats) = match len >= BEACON_TABLE_BYTES {
true => (buf[BEACON_TAKEN_AT], buf[BEACON_SEATS_AT]),
false => (0, 0),
};
Beacon::Here {
id,
name: beacon_name(&buf, len, BEACON_NAME_AT),
host: beacon_name(&buf, len, BEACON_HOST_AT),
taken,
seats,
running: kind == Some(BEACON_RUNNING),
}
}
};
let mut host = from;
host.set_port(game_port);
out.push((host, beacon));
}
}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => break,
Err(_) => break,
}
}
out
}
}
pub struct Announcer {
socket: UdpSocket,
id: u64,
subnet: std::sync::atomic::AtomicU32,
sent: std::sync::atomic::AtomicU32,
}
pub fn subnet_broadcast(ip: std::net::IpAddr) -> Option<std::net::Ipv4Addr> {
let std::net::IpAddr::V4(ip) = ip else {
return None;
};
let [a, b, c, _] = ip.octets();
Some(std::net::Ipv4Addr::new(a, b, c, 255))
}
fn bits_of(addr: Option<std::net::Ipv4Addr>) -> u32 {
addr.map_or(0, u32::from)
}
impl Announcer {
fn subnet(&self, round: u32) -> Option<std::net::Ipv4Addr> {
use std::sync::atomic::Ordering::Relaxed;
if due_for_recheck(round) {
let fresh = bits_of(local_ip().and_then(subnet_broadcast));
self.subnet.store(fresh, Relaxed);
}
match self.subnet.load(Relaxed) {
0 => None,
bits => Some(std::net::Ipv4Addr::from(bits)),
}
}
pub fn new(id: u64) -> io::Result<Announcer> {
let socket = UdpSocket::bind(("0.0.0.0", 0))?;
socket.set_broadcast(true)?;
socket.set_nonblocking(true)?;
let port = u64::from(socket.local_addr()?.port());
Ok(Announcer {
socket,
id: id.rotate_left(16) ^ port,
subnet: std::sync::atomic::AtomicU32::new(bits_of(
local_ip().and_then(subnet_broadcast),
)),
sent: std::sync::atomic::AtomicU32::new(0),
})
}
pub fn announce(&self, game_port: u16, on_air: OnAir<'_>) {
self.beacon(game_port, BEACON_OPEN, on_air, 1);
}
pub fn running(&self, game_port: u16, on_air: OnAir<'_>) {
self.beacon(game_port, BEACON_RUNNING, on_air, 1);
}
pub fn closing(&self, game_port: u16) {
self.beacon(
game_port,
BEACON_CLOSING,
OnAir::default(),
FAREWELL_REPEATS,
);
}
fn beacon(&self, game_port: u16, kind: u8, on_air: OnAir<'_>, times: usize) {
let OnAir {
name,
host,
taken,
seats,
} = on_air;
let mut packet = ANNOUNCE_MAGIC.to_vec();
packet.extend_from_slice(&game_port.to_le_bytes());
packet.push(kind);
packet.extend_from_slice(&wire_name(name));
packet.push(taken);
packet.push(seats);
packet.extend_from_slice(&self.id.to_le_bytes());
packet.extend_from_slice(&wire_name(host));
debug_assert_eq!(packet.len(), BEACON_BYTES);
let round = self.sent.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let wide = wide_ports(round, times > 1);
let subnet = self.subnet(round);
for _ in 0..times {
for port in LOBBY_PORTS {
if wide.contains(&port) {
let _ = self.socket.send_to(&packet, ("255.255.255.255", port));
if let Some(subnet) = subnet {
let _ = self.socket.send_to(&packet, (subnet, port));
}
}
let _ = self.socket.send_to(&packet, ("127.0.0.1", port));
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_network_is_asked_about_again_on_a_cadence() {
let due: Vec<u32> = (0..20).filter(|round| due_for_recheck(*round)).collect();
assert_eq!(due, [8, 16]);
}
#[test]
fn a_subnet_that_went_stale_does_not_outlive_the_recheck() {
use std::sync::atomic::Ordering::Relaxed;
let announcer = Announcer::new(0x5AB1).expect("announcer");
let stale = std::net::Ipv4Addr::new(203, 0, 113, 255);
announcer.subnet.store(u32::from(stale), Relaxed);
assert_eq!(announcer.subnet(1), Some(stale));
assert_eq!(announcer.subnet(SUBNET_RECHECK_EVERY - 1), Some(stale));
assert_ne!(
announcer.subnet(SUBNET_RECHECK_EVERY),
Some(stale),
"a stale network outlived the recheck"
);
}
#[test]
fn only_every_fourth_beacon_shouts_at_the_whole_ladder() {
let wide: Vec<usize> = (0..12).map(|n| wide_ports(n, false).len()).collect();
assert_eq!(wide, [8, 1, 1, 1, 8, 1, 1, 1, 8, 1, 1, 1]);
let sent: usize = wide.iter().take(WIDE_EVERY as usize).sum::<usize>() * 2;
assert_eq!(sent, 22, "per {WIDE_EVERY} seconds");
for round in 0..12 {
assert_eq!(wide_ports(round, true).len(), LOBBY_PORTS.len(), "{round}");
}
}
#[test]
fn the_thinned_beacon_goes_to_the_port_a_lone_listener_holds() {
assert_eq!(wide_ports(1, false), &LOBBY_PORTS[..1]);
}
#[test]
fn a_beacon_also_names_the_network_it_is_on() {
let broadcast =
|ip: &str| subnet_broadcast(ip.parse().expect("addr")).map(|addr| addr.to_string());
assert_eq!(broadcast("192.168.1.8"), Some("192.168.1.255".into()));
assert_eq!(broadcast("10.0.0.1"), Some("10.0.0.255".into()));
assert_eq!(broadcast("10.1.2.3"), Some("10.1.2.255".into()));
assert_eq!(broadcast("192.168.1.255"), Some("192.168.1.255".into()));
assert_eq!(broadcast("::1"), None);
assert_eq!(broadcast("2001:db8::8"), None);
assert_eq!(broadcast("::ffff:10.1.2.3"), None);
}
#[test]
fn discovery_hears_beacons_on_loopback() {
let mut discovery = Discovery::bind().expect("bind lobby port");
let announcer = Announcer::new(0xB0A7).expect("announcer");
let heard = |discovery: &mut Discovery, send: &dyn Fn()| {
let mut found = Vec::new();
for _ in 0..20 {
send();
std::thread::sleep(std::time::Duration::from_millis(10));
found.extend(
discovery
.poll()
.into_iter()
.filter(|(a, _)| a.port() == 48123),
);
if !found.is_empty() {
break;
}
}
found
};
let told = |taken| OnAir {
name: "Room 3",
host: "Anna",
taken,
seats: 6,
};
let open = heard(&mut discovery, &|| announcer.announce(48123, told(3)));
let (host, beacon) = open.first().expect("host discovered");
assert_eq!(host.port(), 48123, "the port the payload named");
let Beacon::Here {
id,
name,
host: whose,
taken,
seats,
running,
} = beacon.clone()
else {
panic!("an open beach, not {beacon:?}")
};
assert_eq!(
(name.as_str(), whose.as_str(), taken, seats, running),
("Room 3", "Anna", 3, 6, false),
"what the beach is called, whose it is and how full: none of \
which an address says"
);
assert_ne!(id, 0, "and which beach, so two are never one row");
let running = heard(&mut discovery, &|| announcer.running(48123, told(5)));
let (_, beacon) = running.first().expect("a running beach still says so");
assert!(
matches!(beacon, Beacon::Here { running: true, .. }),
"{beacon:?}"
);
assert!(beacon.has_room(), "five of six seats is room for one more");
let bye = heard(&mut discovery, &|| announcer.closing(48123));
let (host, beacon) = bye.first().expect("farewell heard");
assert_eq!(host.port(), 48123);
assert!(matches!(beacon, Beacon::Closing { .. }), "{beacon:?}");
}
#[test]
fn two_announcers_from_one_clock_reading_are_still_two() {
let (a, b) = (
Announcer::new(0x1234_5678).expect("one"),
Announcer::new(0x1234_5678).expect("two"),
);
assert_ne!(a.id, b.id, "same reading, different beaches");
}
#[test]
fn every_beacon_fits_the_receive_buffer() {
let mut packet = ANNOUNCE_MAGIC.to_vec();
packet.extend_from_slice(&u16::MAX.to_le_bytes());
packet.push(BEACON_OPEN);
packet.extend_from_slice(&wire_name("WWWWWWWWWWWWWWWWWWWWWWWW"));
packet.push(6);
packet.push(6);
packet.extend_from_slice(&u64::MAX.to_le_bytes());
packet.extend_from_slice(&wire_name("WWWWWWWWWWWWWWWWWWWWWWWW"));
assert!(
packet.len() <= MAX_BEACON,
"{} bytes of beacon, {MAX_BEACON}-byte buffer",
packet.len()
);
assert_eq!(packet.len(), BEACON_BYTES, "the whole layout");
}
#[test]
fn a_beacon_from_before_the_host_had_a_name_keeps_the_rest() {
let mut discovery = Discovery::bind().expect("bind lobby port");
let socket = UdpSocket::bind(("0.0.0.0", 0)).expect("sender");
let mut old = ANNOUNCE_MAGIC.to_vec();
old.extend_from_slice(&48125u16.to_le_bytes());
old.push(BEACON_OPEN);
old.extend_from_slice(&wire_name("Room 3"));
old.push(2);
old.push(6);
old.extend_from_slice(&0x5EA5u64.to_le_bytes());
assert_eq!(old.len(), BEACON_TABLE_BYTES, "the packet as it was");
let mut found = Vec::new();
for _ in 0..20 {
for port in LOBBY_PORTS {
let _ = socket.send_to(&old, ("127.0.0.1", port));
}
std::thread::sleep(std::time::Duration::from_millis(10));
found.extend(
discovery
.poll()
.into_iter()
.filter(|(a, _)| a.port() == 48125),
);
if !found.is_empty() {
break;
}
}
let (_, beacon) = found.first().expect("still a beacon");
assert_eq!(
*beacon,
Beacon::Here {
id: 0x5EA5,
name: "Room 3".to_string(),
host: String::new(),
taken: 2,
seats: 6,
running: false,
},
"everything it said, and nothing invented for what it did not"
);
}
#[test]
fn a_beacon_without_a_kind_byte_still_says_here() {
let mut discovery = Discovery::bind().expect("bind lobby port");
let socket = UdpSocket::bind(("0.0.0.0", 0)).expect("sender");
socket.set_broadcast(true).expect("broadcast");
let mut old = ANNOUNCE_MAGIC.to_vec();
old.extend_from_slice(&48124u16.to_le_bytes());
assert_eq!(old.len(), 7, "the packet as it was before farewells");
let mut found = Vec::new();
for _ in 0..20 {
for port in LOBBY_PORTS {
let _ = socket.send_to(&old, ("127.0.0.1", port));
}
std::thread::sleep(std::time::Duration::from_millis(10));
found.extend(
discovery
.poll()
.into_iter()
.filter(|(a, _)| a.port() == 48124),
);
if !found.is_empty() {
break;
}
}
let (host, beacon) = found.first().expect("an old beacon is still a beacon");
assert_eq!(host.port(), 48124);
assert_eq!(
*beacon,
Beacon::Here {
id: 0,
name: String::new(),
host: String::new(),
taken: 0,
seats: 0,
running: false,
},
"nameless and tableless, listed by address as it always was"
);
assert!(
beacon.has_room(),
"a table it never described is not a full one"
);
}
#[test]
fn a_short_beacon_behind_a_long_one_borrows_none_of_it() {
let mut discovery = Discovery::bind().expect("bind lobby port");
let socket = UdpSocket::bind(("0.0.0.0", 0)).expect("sender");
let mut running = ANNOUNCE_MAGIC.to_vec();
running.extend_from_slice(&48126u16.to_le_bytes());
running.push(BEACON_RUNNING);
running.extend_from_slice(&wire_name("Room 3"));
running.push(2);
running.push(6);
running.extend_from_slice(&0x5EA5u64.to_le_bytes());
running.extend_from_slice(&wire_name("Anna"));
assert_eq!(running.len(), BEACON_BYTES);
let mut old = ANNOUNCE_MAGIC.to_vec();
old.extend_from_slice(&48127u16.to_le_bytes());
assert_eq!(old.len(), 7, "the packet as it was before farewells");
let mut found = Vec::new();
for _ in 0..20 {
for port in LOBBY_PORTS {
let _ = socket.send_to(&running, ("127.0.0.1", port));
let _ = socket.send_to(&old, ("127.0.0.1", port));
}
std::thread::sleep(std::time::Duration::from_millis(10));
found.extend(
discovery
.poll()
.into_iter()
.filter(|(a, _)| a.port() == 48127),
);
if !found.is_empty() {
break;
}
}
let (_, beacon) = found.first().expect("the short beacon is heard");
assert_eq!(
*beacon,
Beacon::Here {
id: 0,
name: String::new(),
host: String::new(),
taken: 0,
seats: 0,
running: false,
},
"an open beach, not the kind of the packet before it"
);
}
}