Partial: Dedicated Peer distribution system

This is an uncompleted commit to move the work over to my other machine. Should compile though.

src/game/net.rs

@@ -2,24 +2,16 @@ 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 {
+pub fn handle_new_peer(new_peers: Query<&Peer, Added<Peer>>) {
     for peer in new_peers {
-            outbound.write(OutboundPacket(Packet::new((*seed).into(), peer.uuid)));
-        }
+        info!("Peer {} was added", peer.id);
     }
 }
 
 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);
+            info!("Peer {} was removed", peer.id);
         } else {
             info!("Peer {} was removed", entity);
         }
@@ -27,12 +19,8 @@ pub fn handle_deleted_peer(mut old_peers: RemovedComponents<Peer>, peers: Query<
     Ok(())
 }
 
-pub fn handle_incoming_packets(mut packets: EventReader<InboundPacket>, mut commands: Commands) {
+pub fn handle_incoming_packets(mut packets: EventReader<InboundPacket>) {
     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);
-        }
+        info!("Packet received: {:?}", packet.0.message);
     }
 }

src/game/plugin.rs

@@ -91,15 +91,17 @@ impl Plugin for GamePlugin {
             ),
         );
 
+        Seed::register(app);
+
         match self.source {
             DataSource::Address(peer) => {
                 info!("Will retrieve seed from peer => {peer}");
             }
             DataSource::Seed(seed) => {
-                app.insert_resource(seed);
+                app.world_mut().spawn(seed);
             }
             DataSource::None => {
-                app.insert_resource(Seed::random());
+                app.world_mut().spawn(Seed::random());
             }
         };
     }

src/game/runtime.rs

@@ -78,6 +78,6 @@ pub fn zoom_camera(
         MouseScrollUnit::Line => 0.1,
         MouseScrollUnit::Pixel => 0.001,
     };
-    projection.scale = (projection.scale - scroll.delta.y * scroll_type_multiplier).clamp(0.1, 2.5);
+    projection.scale = (projection.scale - scroll.delta.y * scroll_type_multiplier).clamp(0.1, 4.0);
     Ok(())
 }

src/game/seed.rs

@@ -7,7 +7,7 @@ use bevy::prelude::*;
 use rand::random;
 
 /// Value with which to initialize the PRNG
-#[derive(Resource, Debug, Clone, Copy)]
+#[derive(Clone, Component, Copy, Debug)]
 pub struct Seed(u64);
 
 impl Seed {
@@ -62,7 +62,6 @@ impl TryFrom<Vec<u8>> for Seed {
     type Error = TryFromSliceError;
 
     fn try_from(value: Vec<u8>) -> Result<Self, Self::Error> {
-        let bytes: [u8; 8] = value.as_slice().try_into()?;
-        Ok(bytes.into())
+        Ok(TryInto::<[u8; 8]>::try_into(value.as_slice())?.into())
     }
 }

src/game/setup.rs

@@ -23,7 +23,7 @@ const BALL_COUNT: u8 = 32;
 const BALL_SIZES: Range<f32> = 10.0..25.0;
 const DIMENSION_SIZES: Range<f32> = 500.0..2000.0;
 
-pub fn check_for_seed(seed: Option<Res<Seed>>, mut next_state: ResMut<NextState<AppState>>) {
+pub fn check_for_seed(seed: Option<Single<&Seed>>, mut next_state: ResMut<NextState<AppState>>) {
     if seed.is_some() {
         next_state.set(AppState::InGame);
     }
@@ -34,8 +34,8 @@ pub fn check_for_seed(seed: Option<Res<Seed>>, mut next_state: ResMut<NextState<
 pub struct PlayableArea(f32, f32);
 
 /// Initialize deterministic values
-pub fn setup_from_seed(mut commands: Commands, seed: Res<Seed>) {
-    let mut rng = WyRand::from_seed((*seed).into());
+pub fn setup_from_seed(mut commands: Commands, seed: Single<&Seed>) {
+    let mut rng = WyRand::from_seed((**seed).into());
     commands.insert_resource(PlayableArea(
         rng.random_range(DIMENSION_SIZES),
         rng.random_range(DIMENSION_SIZES),

src/game/ui.rs

@@ -21,13 +21,10 @@ pub fn setup_seed_ui(mut commands: Commands) {
         .with_child((TextSpan::new("<N/A>"), SeedUI));
 }
 
-pub fn update_seed_ui(seed: Option<Res<Seed>>, text: Query<&mut TextSpan, With<SeedUI>>) {
+pub fn update_seed_ui(seed: Option<Single<&Seed>>, text: Query<&mut TextSpan, With<SeedUI>>) {
     if let Some(value) = seed {
-        if value.is_changed() {
         for mut span in text {
-                let number: u64 = (*value).into();
-                **span = format!("{}", number);
-            }
+            **span = format!("{}", u64::from(**value));
         }
     }
 }
@@ -93,8 +90,8 @@ pub fn update_peer_ui_timings(
         if let Ok((peer, recv, send)) = peers.get(id.0) {
             **row = format!(
                 "{} {:.2} {:.2}",
-                peer.uuid,
-                (time.elapsed() - recv.time().unwrap_or(default())).as_secs_f64(),
+                peer.id,
+                (time.elapsed() - recv.time()).as_secs_f64(),
                 (time.elapsed() - send.time()).as_secs_f64()
             )
         }

src/net/distribution.rs

@@ -0,0 +1,49 @@
+use bevy::prelude::*;
+
+use super::{
+    packet::{InboundPacket, OutboundPacket, Packet},
+    peer::Peer,
+    state::NetworkState,
+};
+
+fn spawner<T: TryFrom<Vec<u8>> + Component>(
+    mut inbound: EventReader<InboundPacket>,
+    mut commands: Commands,
+) {
+    for packet in inbound.read() {
+        if let Ok(entity) = T::try_from(packet.0.message.clone()) {
+            commands.spawn(entity);
+        }
+    }
+}
+
+fn sender<T: Into<Vec<u8>> + Component + Clone>(
+    peers: Query<&Peer, Added<Peer>>,
+    entities: Query<&T>,
+    mut outbound: EventWriter<OutboundPacket>,
+) {
+    for peer in peers {
+        for entity in entities {
+            outbound.write(OutboundPacket(Packet::new(
+                (*entity).clone().into(),
+                peer.id,
+            )));
+        }
+    }
+}
+
+pub trait Networked: Into<Vec<u8>> + TryFrom<Vec<u8>> + Component + Clone {
+    fn register(app: &mut App);
+}
+
+impl<T> Networked for T
+where
+    T: Into<Vec<u8>> + TryFrom<Vec<u8>> + Component + Clone,
+{
+    fn register(app: &mut App) {
+        app.add_systems(
+            FixedUpdate,
+            (sender::<T>, spawner::<T>).run_if(in_state(NetworkState::MultiPlayer)),
+        );
+    }
+}

src/net/io.rs

@@ -3,16 +3,18 @@ use std::time::Duration;
 use bevy::prelude::*;
 use uuid::Uuid;
 
+use crate::net::peer::PotentialPeer;
+
 use super::{
     packet::{InboundPacket, OutboundPacket, Packet},
-    peer::{Peer, PeerChangeEvent, PeerMap, PeerReceiveTiming, PeerSendTiming},
+    peer::{Peer, PeerChangeEvent, PeerData, 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 peers: Query<(&PeerData, &mut PeerReceiveTiming)>,
     mut to_app: EventWriter<InboundPacket>,
     time: Res<Time>,
     mut change_peer: EventWriter<PeerChangeEvent>,
@@ -44,8 +46,7 @@ pub fn handle_network_input(
 pub fn handle_network_output(
     mut from_app: EventReader<OutboundPacket>,
     peer_map: Res<PeerMap>,
-    mut peers: Query<(&Peer, &mut PeerSendTiming)>,
-    config: Res<Config>,
+    mut peers: Query<(&PeerData, &mut PeerSendTiming)>,
     to_socket: Res<NetworkSend>,
     time: Res<Time>,
 ) -> Result {
@@ -53,8 +54,8 @@ pub fn handle_network_output(
         let peer_id = peer_map.try_get(&packet.0.peer)?;
         let (peer, mut last) = peers.get_mut(*peer_id)?;
         // Append our UUID for client identification
-        let message = [packet.0.message.as_slice(), config.id.as_bytes()].concat();
-        to_socket.send(message, peer.addr)?;
+        let message = [packet.0.message.as_slice(), peer.me.as_bytes()].concat();
+        to_socket.send(message, peer.addr.into())?;
         last.update(&time);
     }
     Ok(())
@@ -63,17 +64,24 @@ pub fn handle_network_output(
 const TIMEOUT: Duration = Duration::from_secs(10);
 
 pub fn heartbeat(
-    peers: Query<(&Peer, &PeerSendTiming)>,
+    peers: Query<(AnyOf<(&Peer, &PotentialPeer)>, &PeerSendTiming)>,
     time: Res<Time>,
     mut outbound: EventWriter<OutboundPacket>,
-) {
+) -> Result {
     for (peer, last) in peers {
         // Allow for 2 consecutive missed heartbeats without timing out
         if last.time() + TIMEOUT / 3 > time.elapsed() {
             continue;
         }
-        outbound.write(OutboundPacket(Packet::new(Vec::new(), peer.uuid)));
+        let id = match peer {
+            (None, None) => return Err("No peer identification".into()),
+            (None, Some(potential)) => potential.id,
+            (Some(actual), None) => actual.id,
+            (Some(_), Some(_)) => return Err("Both a potential and actual peer".into()),
+        };
+        outbound.write(OutboundPacket(Packet::new(Vec::new(), id)));
     }
+    Ok(())
 }
 
 pub fn timeout(
@@ -82,22 +90,9 @@ pub fn timeout(
     mut delete: EventWriter<PeerChangeEvent>,
 ) {
     for (peer, last) in peers {
-        if let Some(previous) = last.time() {
-            if previous + TIMEOUT < time.elapsed() {
-                warn!("Peer {} timed out", peer.uuid);
-                delete.write(PeerChangeEvent::new(peer.uuid, None));
-            }
+        if last.time() + TIMEOUT < time.elapsed() {
+            warn!("Peer {} timed out", peer.id);
+            delete.write(PeerChangeEvent::new(peer.id, None));
         }
     }
 }
-
-#[derive(Debug, Resource)]
-pub struct Config {
-    pub id: Uuid,
-}
-
-impl Config {
-    pub fn new() -> Self {
-        Self { id: Uuid::new_v4() }
-    }
-}

src/net/mod.rs

@@ -1,14 +1,18 @@
+mod distribution;
 mod io;
 mod packet;
 mod peer;
 mod plugin;
 mod queues;
 mod socket;
+mod state;
 mod thread;
 
 #[allow(unused_imports)]
 pub mod prelude {
+    pub use super::distribution::Networked;
     pub use super::packet::{InboundPacket, OutboundPacket, Packet};
     pub use super::peer::{Peer, PeerReceiveTiming, PeerSendTiming};
-    pub use super::plugin::{NetIOPlugin, NetworkState};
+    pub use super::plugin::NetIOPlugin;
+    pub use super::state::NetworkState;
 }

src/net/peer.rs

@@ -1,4 +1,9 @@
-use std::{collections::HashMap, net::SocketAddr, time::Duration};
+use std::{
+    array::TryFromSliceError,
+    collections::HashMap,
+    net::{IpAddr, Ipv6Addr, SocketAddr},
+    time::Duration,
+};
 
 use bevy::prelude::*;
 use uuid::Uuid;
@@ -19,32 +24,122 @@ impl PeerSendTiming {
 }
 
 #[derive(Component, Debug, Default)]
-pub struct PeerReceiveTiming(Option<Duration>);
+pub struct PeerReceiveTiming(Duration);
 
 impl PeerReceiveTiming {
     pub fn new(time: &Res<Time>) -> Self {
-        Self(Some(time.elapsed()))
+        Self(time.elapsed())
     }
 
     pub fn update(&mut self, time: &Res<Time>) {
-        self.0 = Some(time.elapsed())
+        self.0 = time.elapsed()
     }
 
-    pub fn time(&self) -> Option<Duration> {
+    pub fn time(&self) -> Duration {
         self.0
     }
 }
 
 #[derive(Component, Debug)]
-#[require(PeerSendTiming, PeerReceiveTiming)]
+#[require(PeerData, PeerReceiveTiming)]
 pub struct Peer {
-    pub addr: SocketAddr,
-    pub uuid: Uuid,
+    pub id: Uuid,
 }
 
-impl Peer {
-    pub fn new(addr: SocketAddr, uuid: Uuid) -> Self {
-        Self { addr, uuid }
+#[derive(Component, Debug)]
+#[require(PeerData)]
+pub struct PotentialPeer {
+    pub id: Uuid,
+}
+
+impl Default for PotentialPeer {
+    fn default() -> Self {
+        Self { id: Uuid::new_v4() }
+    }
+}
+
+#[derive(Clone, Component, Copy, Debug)]
+#[require(PeerSendTiming)]
+pub struct PeerData {
+    pub addr: Address,
+    pub me: Uuid,
+}
+
+impl PeerData {
+    pub fn new(addr: SocketAddr, me: Uuid) -> Self {
+        Self {
+            addr: addr.into(),
+            me,
+        }
+    }
+}
+
+impl Default for PeerData {
+    fn default() -> Self {
+        Self {
+            addr: Default::default(),
+            me: Uuid::new_v4(),
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct Address(SocketAddr);
+
+impl From<Address> for SocketAddr {
+    fn from(value: Address) -> Self {
+        value.0
+    }
+}
+
+impl From<SocketAddr> for Address {
+    fn from(value: SocketAddr) -> Self {
+        Self(value)
+    }
+}
+
+impl PartialEq<SocketAddr> for Address {
+    fn eq(&self, other: &SocketAddr) -> bool {
+        self.0 == *other
+    }
+}
+
+impl PartialEq<Address> for SocketAddr {
+    fn eq(&self, other: &Address) -> bool {
+        other == self
+    }
+}
+
+impl From<Address> for Vec<u8> {
+    fn from(value: Address) -> Self {
+        let mut bytes: Vec<u8> = match value.0.ip() {
+            IpAddr::V4(ipv4_addr) => ipv4_addr.octets().into(),
+            IpAddr::V6(ipv6_addr) => ipv6_addr.octets().into(),
+        };
+        bytes.extend(value.0.port().to_le_bytes());
+        bytes
+    }
+}
+
+impl TryFrom<Vec<u8>> for Address {
+    type Error = TryFromSliceError;
+
+    fn try_from(value: Vec<u8>) -> Result<Self, Self::Error> {
+        let port = u16::from_le_bytes(TryInto::<[u8; 2]>::try_into(
+            value.clone().split_off(value.len() - 2).as_slice(),
+        )?);
+        let addr = if let Ok(bytes) = TryInto::<[u8; 4]>::try_into(value.as_slice()) {
+            SocketAddr::from((bytes, port))
+        } else {
+            SocketAddr::from((TryInto::<[u8; 16]>::try_into(value.as_slice())?, port))
+        };
+        Ok(Address(addr))
+    }
+}
+
+impl Default for Address {
+    fn default() -> Self {
+        Self(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0))
     }
 }
 
@@ -90,7 +185,7 @@ impl PeerChangeEvent {
 pub fn handle_peer_change(
     mut changes: EventReader<PeerChangeEvent>,
     mut peer_map: ResMut<PeerMap>,
-    mut peers: Query<&mut Peer>,
+    mut peers: Query<&mut PeerData>,
     mut commands: Commands,
     time: Res<Time>,
 ) -> Result {
@@ -98,7 +193,7 @@ pub fn handle_peer_change(
         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;
+                    peer.addr = Address(addr);
                 } else {
                     warn!("Peer {} doesn't exist (just added?)", change.peer);
                 }
@@ -111,7 +206,10 @@ pub fn handle_peer_change(
             peer_map.insert(
                 change.peer,
                 commands
-                    .spawn((Peer::new(addr, change.peer), PeerReceiveTiming::new(&time)))
+                    .spawn((
+                        PeerData::new(addr, Uuid::new_v4()),
+                        PeerReceiveTiming::new(&time),
+                    ))
                     .id(),
             );
         } else {
@@ -126,11 +224,25 @@ pub fn handle_peer_change(
     Ok(())
 }
 
+#[allow(clippy::type_complexity)]
 pub fn handle_new_peer(
-    new_peers: Query<&Peer, Added<Peer>>,
+    new_peers: Query<(Option<Ref<Peer>>, Option<&Peer>, &PeerData)>,
     mut outbound: EventWriter<OutboundPacket>,
-) {
-    for peer in new_peers {
-        outbound.write(OutboundPacket(Packet::new(Vec::new(), peer.uuid)));
+) -> Result {
+    for (change, this, data) in new_peers {
+        if let Some(change) = change {
+            if change.is_added() {
+                if let Some(this) = this {
+                    for (_, _, other) in new_peers {
+                        if data.me != other.me {
+                            outbound.write(OutboundPacket(Packet::new(other.addr.into(), this.id)));
                         }
+                    }
+                } else {
+                    return Err("Ref<Peer> without Peer".into());
+                }
+            }
+        }
+    }
+    Ok(())
 }

src/net/plugin.rs

@@ -4,20 +4,14 @@ use bevy::prelude::*;
 use uuid::Uuid;
 
 use super::{
-    io::{Config, handle_network_input, handle_network_output, heartbeat, timeout},
+    io::{handle_network_input, handle_network_output, heartbeat, timeout},
     packet::{InboundPacket, OutboundPacket},
-    peer::{Peer, PeerChangeEvent, PeerMap, handle_new_peer, handle_peer_change},
+    peer::{PeerChangeEvent, PeerData, PeerMap, handle_new_peer, handle_peer_change},
     queues::{NetworkReceive, NetworkSend},
     socket::bind_socket,
+    state::NetworkState,
 };
 
-#[derive(States, Default, Debug, Clone, PartialEq, Eq, Hash)]
-pub enum NetworkState {
-    #[default]
-    SinglePlayer,
-    MultiPlayer,
-}
-
 pub struct NetIOPlugin {
     listen: u16,
     peer: Option<SocketAddr>,
@@ -46,7 +40,6 @@ impl Plugin for NetIOPlugin {
                 FixedPostUpdate,
                 handle_network_output.run_if(in_state(NetworkState::MultiPlayer)),
             )
-            .insert_resource(Config::new())
             .add_event::<PeerChangeEvent>()
             .add_event::<InboundPacket>()
             .add_event::<OutboundPacket>();
@@ -59,7 +52,10 @@ impl Plugin for NetIOPlugin {
 
                 let mut peer_map = PeerMap::default();
                 if let Some(socket) = self.peer {
-                    let entity = app.world_mut().spawn(Peer::new(socket, Uuid::nil()));
+                    let entity = app.world_mut().spawn(PeerData {
+                        addr: socket.into(),
+                        me: Uuid::nil(),
+                    });
                     peer_map.insert(Uuid::nil(), entity.id());
                 }
 

src/net/state.rs

@@ -0,0 +1,8 @@
+use bevy::prelude::*;
+
+#[derive(States, Default, Debug, Clone, PartialEq, Eq, Hash)]
+pub enum NetworkState {
+    #[default]
+    SinglePlayer,
+    MultiPlayer,
+}