Rework networking interface

Add Bevy event queues for inbound and outbound packets, and use Bevy change detection for consumer new Peer handling.

Cargo.lock

@@ -1924,6 +1924,7 @@ dependencies = [
  "crossbeam-channel",
  "log",
  "rand 0.9.1",
+ "uuid",
  "wyrand",
 ]
 
@@ -4479,9 +4480,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
 
 [[package]]
 name = "uuid"
-version = "1.16.0"
+version = "1.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "458f7a779bf54acc9f347480ac654f68407d3aab21269a6e3c9f922acd9e2da9"
+checksum = "3cf4199d1e5d15ddd86a694e4d0dffa9c323ce759fea589f00fef9d81cc1931d"
 dependencies = [
  "getrandom 0.3.3",
  "js-sys",

Cargo.toml

@@ -42,6 +42,7 @@ rand = { version = "0.9.1", default-features = false, features = [
 	"std",
 	"thread_rng",
 ] }
+uuid = { version = "1.17.0", features = ["v4"] }
 wyrand = "0.3.2"
 
 [features]

src/game/mod.rs

@@ -1,3 +1,4 @@
+mod net;
 mod objects;
 mod plugin;
 mod rng;

src/game/net.rs

@@ -0,0 +1,39 @@
+use bevy::prelude::*;
+
+use crate::net::prelude::*;
+
+use super::seed::Seed;
+
+pub fn handle_new_peer(
+    seed: Option<Res<Seed>>,
+    new_peers: Query<&Peer, Added<Peer>>,
+    mut outbound: EventWriter<OutboundPacket>,
+) {
+    if let Some(seed) = seed {
+        for peer in new_peers {
+            warn!("Sending seed to peer: {}", peer.uuid);
+            outbound.write(OutboundPacket(Packet::new((*seed).into(), peer.uuid)));
+        }
+    }
+}
+
+pub fn handle_deleted_peer(mut old_peers: RemovedComponents<Peer>, peers: Query<&Peer>) -> Result {
+    for entity in old_peers.read() {
+        if let Ok(peer) = peers.get(entity) {
+            info!("Peer {} was removed", peer.uuid);
+        } else {
+            info!("Peer {} was removed", entity);
+        }
+    }
+    Ok(())
+}
+
+pub fn handle_incoming_packets(mut packets: EventReader<InboundPacket>, mut commands: Commands) {
+    for packet in packets.read() {
+        if let Ok(seed) = packet.0.message.clone().try_into() {
+            commands.insert_resource::<Seed>(seed);
+        } else {
+            info!("Packet not seed: {:?}", packet.0.message);
+        }
+    }
+}

src/game/plugin.rs

@@ -3,9 +3,10 @@ use std::net::SocketAddr;
 use avian2d::PhysicsPlugins;
 use bevy::{input::common_conditions::input_pressed, prelude::*};
 
-use crate::net::NetIOPlugin;
+use crate::net::prelude::*;
 
 use super::{
+    net::{handle_deleted_peer, handle_incoming_packets, handle_new_peer},
     runtime::{move_camera, move_player, quit, zoom_camera},
     seed::Seed,
     setup::{check_for_seed, setup_balls, setup_from_seed, setup_player, setup_ui, setup_walls},
@@ -65,7 +66,15 @@ impl Plugin for GamePlugin {
         )
         .add_systems(
             FixedUpdate,
+            (
                 check_for_seed.run_if(in_state(AppState::Loading)),
+                (
+                    handle_new_peer,
+                    handle_deleted_peer,
+                    handle_incoming_packets,
+                )
+                    .run_if(in_state(NetworkState::MultiPlayer)),
+            ),
         )
         .add_systems(
             Update,

src/net/io.rs

@@ -0,0 +1,118 @@
+use std::time::Duration;
+
+use bevy::prelude::*;
+use uuid::Uuid;
+
+use super::{
+    peer::{Peer, PeerChangeEvent, PeerMap, PeerReceiveTiming, PeerSendTiming},
+    queues::{NetworkReceive, NetworkSend},
+};
+
+pub fn handle_network_input(
+    from_socket: Res<NetworkReceive>,
+    peer_map: Res<PeerMap>,
+    mut peers: Query<(&Peer, &mut PeerReceiveTiming)>,
+    mut to_app: EventWriter<InboundPacket>,
+    time: Res<Time>,
+    mut change_peer: EventWriter<PeerChangeEvent>,
+) -> Result {
+    for (mut message, address) in from_socket.iter() {
+        if message.len() < 16 {
+            return Err(format!("Message of length {} cannot contain UUID", message.len()).into());
+        }
+        let uuid = Uuid::from_slice(message.split_off(message.len() - 16).as_slice())?;
+        info!("Receiving: {:?} from {}", message, uuid);
+        to_app.write(InboundPacket(Packet::new(message, uuid)));
+        if let Some(peer_id) = peer_map.get(&uuid) {
+            let (peer, mut last) = peers.get_mut(*peer_id)?;
+            last.update(&time);
+            if address == peer.addr {
+                continue;
+            }
+        }
+        change_peer.write(PeerChangeEvent::new(uuid, Some(address)));
+    }
+    Ok(())
+}
+
+pub fn handle_network_output(
+    mut from_app: EventReader<OutboundPacket>,
+    peer_map: Res<PeerMap>,
+    mut peers: Query<(&Peer, &mut PeerSendTiming)>,
+    config: Res<Config>,
+    to_socket: Res<NetworkSend>,
+    time: Res<Time>,
+) -> Result {
+    for packet in from_app.read() {
+        let peer_id = peer_map.try_get(&packet.0.peer)?;
+        let (peer, mut last) = peers.get_mut(*peer_id)?;
+        warn!("Sending: {:?} to {}", packet.0.message, peer.uuid);
+        // Append our UUID for client identification
+        let message = [packet.0.message.as_slice(), config.id.as_bytes()].concat();
+        to_socket.send(message, peer.addr)?;
+        last.update(&time);
+    }
+    Ok(())
+}
+
+const TIMEOUT: Duration = Duration::from_secs(10);
+
+pub fn heartbeat(
+    peers: Query<(&Peer, &PeerSendTiming)>,
+    time: Res<Time>,
+    mut outbound: EventWriter<OutboundPacket>,
+) {
+    for (peer, last) in peers {
+        if let Some(previous) = last.timestamp() {
+            // Allow for 2 consecutive missed heartbeats without timing out
+            if previous + TIMEOUT / 3 > time.elapsed() {
+                continue;
+            }
+        }
+        outbound.write(OutboundPacket(Packet::new(Vec::new(), peer.uuid)));
+    }
+}
+
+pub fn timeouts(
+    peers: Query<(&Peer, &PeerReceiveTiming)>,
+    time: Res<Time>,
+    mut delete: EventWriter<PeerChangeEvent>,
+) {
+    for (peer, last) in peers {
+        if let Some(previous) = last.timestamp() {
+            if previous + TIMEOUT < time.elapsed() {
+                warn!("Peer {} timed out", peer.uuid);
+                delete.write(PeerChangeEvent::new(peer.uuid, None));
+            }
+        }
+    }
+}
+
+#[derive(Debug, Resource)]
+pub struct Config {
+    pub id: Uuid,
+}
+
+impl Config {
+    pub fn new() -> Self {
+        Self { id: Uuid::new_v4() }
+    }
+}
+
+#[derive(Debug)]
+pub struct Packet {
+    pub message: Vec<u8>,
+    pub peer: Uuid,
+}
+
+impl Packet {
+    pub fn new(message: Vec<u8>, peer: Uuid) -> Self {
+        Self { peer, message }
+    }
+}
+
+#[derive(Debug, Event)]
+pub struct OutboundPacket(pub Packet);
+
+#[derive(Debug, Event)]
+pub struct InboundPacket(pub Packet);

src/net/mod.rs

@@ -1,5 +1,13 @@
+mod io;
+mod peer;
 mod plugin;
-pub use plugin::NetIOPlugin;
 mod queues;
 mod socket;
 mod thread;
+
+#[allow(unused_imports)]
+pub mod prelude {
+    pub use super::io::{InboundPacket, OutboundPacket, Packet};
+    pub use super::peer::Peer;
+    pub use super::plugin::{NetIOPlugin, NetworkState};
+}

src/net/peer.rs

@@ -0,0 +1,134 @@
+use std::{collections::HashMap, net::SocketAddr, time::Duration};
+
+use bevy::prelude::*;
+use uuid::Uuid;
+
+#[derive(Debug, Default)]
+struct Timing {
+    timestamp: Option<Duration>,
+}
+
+#[derive(Component, Debug, Default)]
+pub struct PeerSendTiming(Timing);
+
+impl PeerSendTiming {
+    pub fn update(&mut self, time: &Res<Time>) {
+        self.0.timestamp = Some(time.elapsed())
+    }
+
+    pub fn timestamp(&self) -> Option<Duration> {
+        self.0.timestamp
+    }
+}
+
+#[derive(Component, Debug, Default)]
+pub struct PeerReceiveTiming(Timing);
+
+impl PeerReceiveTiming {
+    pub fn new(time: &Res<Time>) -> Self {
+        Self(Timing {
+            timestamp: Some(time.elapsed()),
+        })
+    }
+
+    pub fn update(&mut self, time: &Res<Time>) {
+        self.0.timestamp = Some(time.elapsed())
+    }
+
+    pub fn timestamp(&self) -> Option<Duration> {
+        self.0.timestamp
+    }
+}
+
+#[derive(Component, Debug)]
+#[require(PeerSendTiming, PeerReceiveTiming)]
+pub struct Peer {
+    pub addr: SocketAddr,
+    pub uuid: Uuid,
+}
+
+impl Peer {
+    pub fn new(addr: SocketAddr, uuid: Uuid) -> Self {
+        Self { addr, uuid }
+    }
+}
+
+#[derive(Debug, Default, Resource)]
+pub struct PeerMap {
+    map: HashMap<Uuid, Entity>,
+}
+
+impl PeerMap {
+    pub fn insert(&mut self, uuid: Uuid, id: Entity) {
+        self.map.insert(uuid, id);
+    }
+
+    pub fn get(&self, uuid: &Uuid) -> Option<&Entity> {
+        self.map.get(uuid)
+    }
+
+    pub fn try_get(&self, uuid: &Uuid) -> Result<&Entity, String> {
+        self.get(uuid).ok_or(format!("No entity with uuid: {uuid}"))
+    }
+
+    pub fn remove(&mut self, uuid: &Uuid) -> Option<Entity> {
+        self.map.remove(uuid)
+    }
+
+    pub fn len(&self) -> usize {
+        self.map.len()
+    }
+}
+
+#[derive(Debug, Event)]
+pub struct PeerChangeEvent {
+    peer: Uuid,
+    addr: Option<SocketAddr>,
+}
+
+impl PeerChangeEvent {
+    pub fn new(peer: Uuid, addr: Option<SocketAddr>) -> Self {
+        Self { peer, addr }
+    }
+}
+
+pub fn handle_peer_change(
+    mut changes: EventReader<PeerChangeEvent>,
+    mut peer_map: ResMut<PeerMap>,
+    mut peers: Query<&mut Peer>,
+    mut commands: Commands,
+    time: Res<Time>,
+) -> Result {
+    for change in changes.read() {
+        if let Some(entity) = peer_map.get(&change.peer) {
+            if let Some(addr) = change.addr {
+                if let Ok(mut peer) = peers.get_mut(*entity) {
+                    peer.addr = addr;
+                } else {
+                    warn!("Peer {} doesn't exist (just added?)", change.peer);
+                }
+            } else {
+                info!("Removing peer {}", change.peer);
+                commands.get_entity(*entity)?.despawn();
+                peer_map.remove(&change.peer);
+            }
+        } else if let Some(addr) = change.addr {
+            info!("Adding peer {} ({:?})", change.peer, change.addr);
+            peer_map.insert(
+                change.peer,
+                commands
+                    .spawn((Peer::new(addr, change.peer), PeerReceiveTiming::new(&time)))
+                    .id(),
+            );
+        } else {
+            warn!("Peer {} already deleted", change.peer);
+        }
+    }
+    if peer_map.len() > 1 {
+        if let Some(entity) = peer_map.remove(&Uuid::nil()) {
+            warn!("Removing initial Peer");
+            commands.get_entity(entity)?.despawn();
+        }
+    }
+    Ok(())
+}

src/net/plugin.rs

@@ -1,79 +1,25 @@
-use std::{net::SocketAddr, time::Duration};
+use std::net::SocketAddr;
 
 use bevy::prelude::*;
+use uuid::Uuid;
 
 use super::{
+    io::{
+        Config, InboundPacket, OutboundPacket, handle_network_input, handle_network_output,
+        heartbeat, timeouts,
+    },
+    peer::{Peer, PeerChangeEvent, PeerMap, handle_peer_change},
     queues::{NetworkReceive, NetworkSend},
     socket::bind_socket,
 };
-use crate::game::prelude::Seed;
-
-fn handle_network_input(
-    receive: Res<NetworkReceive>,
-    send: Res<NetworkSend>,
-    seed: Option<Res<Seed>>,
-    mut commands: Commands,
-) -> Result {
-    for (message, address) in receive.iter() {
-        // Temporary logic just for initial connection, if there is already a seed then the peer wants it
-        if let Some(ref value) = seed {
-            send.send((**value).into(), address)?;
-        } else {
-            commands.insert_resource::<Seed>(message.try_into()?);
-        }
-    }
-
-    Ok(())
-}
-
-const TIMEOUT: Duration = Duration::from_secs(5);
-
-fn heartbeat(
-    peers: Query<(&Peer, &mut PeerSendTiming)>,
-    send: Res<NetworkSend>,
-    time: Res<Time>,
-) -> Result {
-    for (peer, ref mut last) in peers {
-        if let Some(previous) = last.timestamp {
-            if previous + TIMEOUT > time.elapsed() {
-                continue;
-            }
-        }
-        send.send(Vec::new(), peer.addr)?;
-        last.timestamp = Some(time.elapsed());
-    }
-    Ok(())
-}
 
 #[derive(States, Default, Debug, Clone, PartialEq, Eq, Hash)]
-enum NetworkState {
+pub enum NetworkState {
     #[default]
     SinglePlayer,
     MultiPlayer,
 }
 
-#[derive(Component, Debug, Default)]
-struct PeerSendTiming {
-    timestamp: Option<Duration>,
-}
-
-#[derive(Component, Debug, Default)]
-struct PeerReceiveTiming {
-    timestamp: Option<Duration>,
-}
-
-#[derive(Component, Debug)]
-#[require(PeerSendTiming, PeerReceiveTiming)]
-struct Peer {
-    addr: SocketAddr,
-}
-
-impl Peer {
-    fn new(addr: SocketAddr) -> Self {
-        Self { addr }
-    }
-}
-
 pub struct NetIOPlugin {
     listen: u16,
     peer: Option<SocketAddr>,
@@ -87,10 +33,25 @@ impl NetIOPlugin {
 
 impl Plugin for NetIOPlugin {
     fn build(&self, app: &mut App) {
-        app.init_state::<NetworkState>().add_systems(
+        app.init_state::<NetworkState>()
+            .add_systems(
+                FixedPreUpdate,
+                (handle_network_input, handle_peer_change)
+                    .chain()
+                    .run_if(in_state(NetworkState::MultiPlayer)),
+            )
+            .add_systems(
                 FixedUpdate,
-            (handle_network_input, heartbeat).run_if(in_state(NetworkState::MultiPlayer)),
+                (heartbeat, timeouts).run_if(in_state(NetworkState::MultiPlayer)),
-        );
+            )
+            .add_systems(
+                FixedPostUpdate,
+                handle_network_output.run_if(in_state(NetworkState::MultiPlayer)),
+            )
+            .insert_resource(Config::new())
+            .add_event::<PeerChangeEvent>()
+            .add_event::<InboundPacket>()
+            .add_event::<OutboundPacket>();
 
         match bind_socket(self.listen) {
             Ok((send, receive)) => {
@@ -98,9 +59,13 @@ impl Plugin for NetIOPlugin {
                     .insert_resource(NetworkSend::new(send))
                     .insert_resource(NetworkReceive::new(receive));
 
+                let mut peer_map = PeerMap::default();
                 if let Some(socket) = self.peer {
-                    app.world_mut().spawn(Peer::new(socket));
+                    let entity = app.world_mut().spawn(Peer::new(socket, Uuid::nil()));
+                    peer_map.insert(Uuid::nil(), entity.id());
                 }
+
+                app.insert_resource(peer_map);
             }
             Err(err) => {
                 warn!("Failed to set up networking: {err}");