From aa66f9e7ccc943f231bba5bdd9d58d46d935a056 Mon Sep 17 00:00:00 2001 From: ghostly_zsh Date: Sat, 6 Jun 2026 15:41:27 -0500 Subject: [PATCH] working on making the game work with this networking --- crates/unified/src/client/mod.rs | 31 +++++- crates/unified/src/client/net.rs | 20 ++++ crates/unified/src/server/mod.rs | 51 ++++++---- crates/unified/src/server/net.rs | 18 ++++ crates/unified/src/server/player/join.rs | 3 +- crates/unified/src/shared/ecs.rs | 2 + crates/unified/src/shared/net.rs | 114 +++++++++++++++++++---- 7 files changed, 201 insertions(+), 38 deletions(-) create mode 100644 crates/unified/src/client/net.rs create mode 100644 crates/unified/src/server/net.rs diff --git a/crates/unified/src/client/mod.rs b/crates/unified/src/client/mod.rs index 22679872e2f15f2314a21fb879e60eb2cd369dbe..337c0b250daa62ff6026600c69a11300ef93c0d6 100644 --- a/crates/unified/src/client/mod.rs +++ b/crates/unified/src/client/mod.rs @@ -1,5 +1,6 @@ use aeronet::io::{Session, SessionEndpoint}; use aeronet::io::connection::{DisconnectReason, Disconnected}; +use aeronet_transport::lane::LaneKind; //use aeronet_replicon::client::AeronetRepliconClient; use aeronet_transport::{Transport, TransportConfig}; use aeronet_websocket::client::{ClientConfig, WebSocketClient}; @@ -21,7 +22,7 @@ use crate::client::components::Me; use crate::client::ship::attachment::client_attachment_plugin; use crate::shared::ecs::{GameplayState, TimeOffset}; use crate::shared::gravity::update_gravity; -use crate::shared::net::Hi; +use crate::shared::net::{Hi, ClientMessageRegistry, ServerMessageRegistry}; use crate::shared::orbit::OrbitPlugin; pub mod colors; @@ -38,6 +39,7 @@ pub mod starguide; pub mod crafting; pub mod components; pub mod plugins; +mod net; pub struct ClientPlugin { pub server: Option @@ -47,6 +49,7 @@ impl Plugin for ClientPlugin { fn build(&self, app: &mut App) { app .init_gizmo_group::() + .add_plugins(net::net_plugin) .add_plugins(rendering::render_plugin) .add_plugins(OrbitPlugin) .add_systems(FixedUpdate, update_gravity.before(PhysicsSystems::Prepare)) @@ -96,9 +99,33 @@ pub fn on_connecting( let name = names.get(entity).unwrap(); info!("{name} is connecting"); } -pub fn on_connected(trigger: On, names: Query<&Name>) { +pub fn on_connected( + trigger: On, + names: Query<&Name>, + sessions: Query<&Session>, + server_message_registry: Res, + client_message_registry: Res, + mut commands: Commands, +) { let entity = trigger.event_target(); let name = names.get(entity).unwrap(); + let Ok(session) = sessions.get(entity) else { + return; + }; + + let server_message_count = server_message_registry.message_count(); + let client_message_count = client_message_registry.message_count(); + let message_count = server_message_count + client_message_count; + debug!("message count: {}", message_count); + let lanes = [LaneKind::ReliableOrdered].repeat(message_count); + let transport = Transport::new( + session, + lanes.clone(), + lanes, + bevy::platform::time::Instant::now(), + ).expect("packet MTU too small to support transport"); + commands.entity(entity).insert(transport); + info!("{name} is connected"); } pub fn on_disconnected(trigger: On, names: Query<&Name>) { diff --git a/crates/unified/src/client/net.rs b/crates/unified/src/client/net.rs new file mode 100644 index 0000000000000000000000000000000000000000..7094a80012e1e607c7016c4a3e9f1cdaaba925c0 --- /dev/null +++ b/crates/unified/src/client/net.rs @@ -0,0 +1,20 @@ +use crate::{prelude::*, shared::net::{ServerEntityMap, SpawnEntity}}; + +pub fn net_plugin(app: &mut App) { + app + .insert_resource(ServerEntityMap::default()) + .add_systems(PostUpdate, spawn_server_entities); +} + +fn spawn_server_entities( + mut spawn_entity_messages: MessageReader, + mut entity_mapper: ResMut, + mut commands: Commands, +) { + for spawn_entity in spawn_entity_messages.read() { + let entity = commands.spawn_empty(); + debug!("client: {:?}, server: {:?}", entity.id(), spawn_entity.server); + entity_mapper.server_to_client.set_mapped(spawn_entity.server, entity.id()); + entity_mapper.client_to_server.set_mapped(entity.id(), spawn_entity.server); + } +} diff --git a/crates/unified/src/server/mod.rs b/crates/unified/src/server/mod.rs index d3ade7ce586732ff6591db079f00b33c9b35d0fb..f4897c574e4d7410d65be116965af821c5305bd3 100644 --- a/crates/unified/src/server/mod.rs +++ b/crates/unified/src/server/mod.rs @@ -5,6 +5,7 @@ mod heat; mod drill; mod craft; mod damping; +pub mod net; pub mod planets; pub mod player; mod system_sets; @@ -28,6 +29,7 @@ use crate::server::damping::damping_plugin; use crate::server::drill::drill_plugin; use crate::server::earth_parts::spawn_parts_plugin; use crate::server::gravity::newtonian_gravity_plugin; +use crate::server::net::net_plugin; use crate::server::part::part_management_plugin; use crate::server::planets::planets_plugin; use crate::server::player::player_management_plugin; @@ -35,6 +37,7 @@ use crate::server::system_sets::{PlayerInputSet, WorldUpdateSet}; use crate::prelude::*; use crate::server::orbit::OrbitPlugin; use crate::server::player::thrust::server_thrust_plugin; +use crate::shared::net::{ClientMessageRegistry, ServerMessageRegistry}; pub struct ServerPlugin { pub bind: SocketAddr @@ -44,6 +47,7 @@ impl Plugin for ServerPlugin { fn build(&self, app: &mut App) { let bind = self.bind; app + .add_plugins(net_plugin) .add_plugins(planets_plugin) .add_plugins(newtonian_gravity_plugin) .add_plugins(player_management_plugin) @@ -91,34 +95,43 @@ fn on_connected( trigger: On, clients: Query<&ChildOf>, sessions: Query<&Session>, + server_message_registry: Res, + client_message_registry: Res, mut commands: Commands, ) { let client = trigger.event_target(); let Ok(&ChildOf(server)) = clients.get(client) else { return; }; + let Ok(session) = sessions.get(client) else { + return; + }; info!(?client, ?server, "client connected"); -} + let player = commands + .spawn((ConnectedGameEntity { + network_entity: client, + })) + .id(); -/*fn handle_authorized( - newly_authorized_clients: Query>, - mut commands: Commands -) { - for client in newly_authorized_clients.iter() { - let player = commands - .spawn((Replicated, ConnectedGameEntity { - network_entity: client, - })) - .id(); + let server_message_count = server_message_registry.message_count(); + let client_message_count = client_message_registry.message_count(); + let message_count = server_message_count + client_message_count; + debug!("message count: {}", message_count); + let lanes = [LaneKind::ReliableOrdered].repeat(message_count); + let transport = Transport::new( + session, + lanes.clone(), + lanes, + bevy::platform::time::Instant::now(), + ).expect("packet MTU too small to support transport"); + commands.entity(client).insert(( + ConnectedNetworkEntity { + game_entity: player, + }, + transport + )); +} - commands.entity(client).insert(( - Replicated, - ConnectedNetworkEntity { - game_entity: player, - }, - )); - } -}*/ fn on_disconnected( trigger: On, clients: Query<&ChildOf>, diff --git a/crates/unified/src/server/net.rs b/crates/unified/src/server/net.rs new file mode 100644 index 0000000000000000000000000000000000000000..3664b6c72cceefb2cf8a88ce12c0eab1ffcbafd3 --- /dev/null +++ b/crates/unified/src/server/net.rs @@ -0,0 +1,18 @@ +use crate::{prelude::*, shared::net::{Mapped, SendTargets, SpawnEntity, ToClients}}; + +pub fn net_plugin(app: &mut App) { + app.add_systems(PreUpdate, detect_entity_spawn); +} + +fn detect_entity_spawn( + mapped_entities: Query>, + mut spawn_entity: MessageWriter>, +) { + for entity in mapped_entities { + debug!("entity was spawned"); + spawn_entity.write(ToClients { + message: SpawnEntity { server: entity }, + targets: SendTargets::All, + }); + } +} diff --git a/crates/unified/src/server/player/join.rs b/crates/unified/src/server/player/join.rs index 3d43e5051cbfebf540dea261f9d12bbeddd6d22e..df6aeda1ab34b866b33e65f0bb20df2b99d438de 100644 --- a/crates/unified/src/server/player/join.rs +++ b/crates/unified/src/server/player/join.rs @@ -5,7 +5,7 @@ use crate::shared::ecs::{Player, PlayerStorage}; use crate::prelude::*; use crate::server::ConnectedGameEntity; use crate::server::part::SpawnPartRequest; -use crate::shared::net::{ClientId, Hi, SendTargets, ToClients}; +use crate::shared::net::{ClientId, Hi, SendTargets, ToClients, Mapped}; use crate::shared::world_config::WorldConfigResource; const SPAWN_ORBIT_OFFSET: f64 = 150.0; @@ -60,6 +60,7 @@ fn join_player(joined_player: Entity, mut commands: Commands, wc: &GlobalWorldCo } #[derive(Component)] +#[require(Mapped)] pub struct PendingPlayer; pub fn handle_new_players( diff --git a/crates/unified/src/shared/ecs.rs b/crates/unified/src/shared/ecs.rs index fcda652bedc9fa7ecceb9b0a344a2f3cbdf7c8a9..37ba0686121d97f9d4173a840a368ec416e36cc4 100644 --- a/crates/unified/src/shared/ecs.rs +++ b/crates/unified/src/shared/ecs.rs @@ -1,5 +1,6 @@ pub mod thruster; +use crate::shared::net::Mapped; use crate::shared::config::part::PartConfig; use bevy::math::{Quat, Vec2}; use bevy::camera::visibility::RenderLayers; @@ -40,6 +41,7 @@ pub struct PartHandle(pub Handle); pub struct TimeOffset(pub f64); #[derive(Component, Serialize, Deserialize, Debug)] +#[require(Mapped)] pub struct Player { pub client: Entity, } diff --git a/crates/unified/src/shared/net.rs b/crates/unified/src/shared/net.rs index cfcd9f870d1be814dbadec3f7f2d2daa7ddcbee7..a927be617970a9763253a816e9e074c65ffc2513 100644 --- a/crates/unified/src/shared/net.rs +++ b/crates/unified/src/shared/net.rs @@ -7,9 +7,10 @@ use aeronet_transport::Transport; use aeronet_transport::lane::LaneIndex; use aeronet_websocket::tungstenite::Bytes; use avian2d::prelude::{AngularInertia, AngularVelocity, Collider, LinearVelocity, Mass, Position, Rotation}; -use bevy::ecs::entity::MapEntities; +use bevy::ecs::entity::{EntityHashMap, MapEntities}; use bevy::ecs::system::SystemState; use bevy::prelude::*; +use bevy::reflect::DynamicTypePath; use postcard::{from_bytes, to_allocvec, to_slice, to_vec}; use crate::prelude::{App, Message}; use crate::shared::thrust::ThrustSolution; @@ -61,7 +62,8 @@ use crate::shared::ecs::thruster::{Thruster, ThrusterOfPart}; pub fn register_net(app: &mut App) { app - .add_server_message::() + .add_mapped_server_message::() + .add_server_message::() .add_client_message::() .add_client_message::() @@ -69,18 +71,22 @@ pub fn register_net(app: &mut App) { .add_client_message::(); } -#[derive(Message, Deserialize, Serialize, TypePath, MapEntities)] +#[derive(Message, Clone, Deserialize, Serialize, TypePath, MapEntities)] pub struct Hi { #[entities] pub you_are: Entity, pub time_offset: f64, } +#[derive(Message, Deserialize, Serialize, TypePath)] +pub struct SpawnEntity { + pub server: Entity, +} pub fn setup_net(app: &mut App) { app.insert_resource(ServerMessageRegistry::default()); app.insert_resource(ClientMessageRegistry::default()); - app.add_systems(PostUpdate, recv_from_server); - app.add_systems(PostUpdate, recv_from_client); + app.add_systems(Update, recv_from_server); + app.add_systems(Update, recv_from_client); } #[derive(Message, Deref, Deserialize, Serialize)] pub struct ToClients { @@ -106,17 +112,34 @@ pub enum ClientId { Server, Client(Entity), } +#[derive(Component, Default)] +pub struct Mapped; +#[derive(Resource, Default)] +pub struct ServerEntityMap { + pub server_to_client: EntityHashMap, + pub client_to_server: EntityHashMap, +} #[derive(Resource, Default)] pub struct ServerMessageRegistry { forward: HashMap, reverse: HashMap, &mut World)>, } +impl ServerMessageRegistry { + pub fn message_count(&self) -> usize { + self.forward.len() + } +} #[derive(Resource, Default)] pub struct ClientMessageRegistry { forward: HashMap, reverse: HashMap, ClientId, &mut World)>, } +impl ClientMessageRegistry { + pub fn message_count(&self) -> usize { + self.forward.len() + } +} static COUNTER: AtomicU32 = AtomicU32::new(0); fn get_lane_index() -> LaneIndex { @@ -124,35 +147,68 @@ fn get_lane_index() -> LaneIndex { } pub trait NetAppExt { + fn add_mapped_server_message Deserialize<'a>>(&mut self) -> &mut Self; fn add_server_message Deserialize<'a>>(&mut self) -> &mut Self; + fn add_mapped_client_message Deserialize<'a>>(&mut self) -> &mut Self; fn add_client_message Deserialize<'a>>(&mut self) -> &mut Self; } impl NetAppExt for App { + fn add_mapped_server_message Deserialize<'a>>(&mut self) -> &mut Self { + let mut registry = self.world_mut().resource_mut::(); + register_mapped_server_message::(&mut registry); + self + .add_message::>() + .add_message::() + .add_systems(Update, send_to_client::) + } fn add_server_message Deserialize<'a>>(&mut self) -> &mut Self { + let mut registry = self.world_mut().resource_mut::(); + register_server_message::(&mut registry); self .add_message::>() .add_message::() - .add_systems(Startup, register_server_message::) .add_systems(Update, send_to_client::) } + fn add_mapped_client_message Deserialize<'a>>(&mut self) -> &mut Self { + let mut registry = self.world_mut().resource_mut::(); + register_client_message::(&mut registry); + self + .add_message::>() + .add_message::() + .add_systems(Update, send_mapped_to_server::) + } fn add_client_message Deserialize<'a>>(&mut self) -> &mut Self { + let mut registry = self.world_mut().resource_mut::(); + register_client_message::(&mut registry); self .add_message::>() .add_message::() - .add_systems(Startup, register_client_message::) .add_systems(Update, send_to_server::) } } -fn register_server_message Deserialize<'a>>(mut registry: ResMut) { - registry.forward.insert(T::type_path().to_string(), get_lane_index()); - registry.reverse.insert(get_lane_index(), |payload: Vec, world: &mut World| { +fn register_mapped_server_message Deserialize<'a>>(registry: &mut ServerMessageRegistry) { + let lane_index = get_lane_index(); + registry.forward.insert(T::type_path().to_string(), lane_index); + registry.reverse.insert(lane_index, |payload: Vec, world: &mut World| { + let mut entity_map = world.resource_mut::(); + let mut message = from_bytes::(&payload).expect(&format!("Failed to deserialize message of type {}", T::type_path())); + message.map_entities(&mut entity_map.server_to_client); + debug!("entity map: {:?}", entity_map.server_to_client); + world.write_message(message).expect("Could not send message to game"); + }); +} +fn register_server_message Deserialize<'a>>(registry: &mut ServerMessageRegistry) { + let lane_index = get_lane_index(); + registry.forward.insert(T::type_path().to_string(), lane_index); + registry.reverse.insert(lane_index, |payload: Vec, world: &mut World| { let message = from_bytes::(&payload).expect(&format!("Failed to deserialize message of type {}", T::type_path())); world.write_message(message).expect("Could not send message to game"); }); } -fn register_client_message Deserialize<'a>>(mut registry: ResMut) { - registry.forward.insert(T::type_path().to_string(), get_lane_index()); - registry.reverse.insert(get_lane_index(), |payload: Vec, client_id: ClientId, world: &mut World| { +fn register_client_message Deserialize<'a>>(registry: &mut ClientMessageRegistry) { + let lane_index = get_lane_index(); + registry.forward.insert(T::type_path().to_string(), lane_index); + registry.reverse.insert(lane_index, |payload: Vec, client_id: ClientId, world: &mut World| { let message = from_bytes::(&payload).expect(&format!("Failed to deserialize message of type {}", T::type_path())); world.write_message(FromClients { client_id, @@ -170,22 +226,27 @@ fn send_to_client( match &message.targets { SendTargets::All => { for (client_entity, mut transport) in &mut clients { + debug!("lane: {:?}", message_registry.forward.get(M::type_path()).unwrap()); + debug!("registry: {:?}", message_registry.forward); transport.send.push( *message_registry.forward.get(M::type_path()) .expect("Failed to get message lane; the message likely isn't serialized yet"), to_allocvec(&message).expect("Failed to serialize message").into(), - Instant::now(), + bevy::platform::time::Instant::now(), ).expect("Failed to send message"); } } SendTargets::Single(client) => { for (client_entity, mut transport) in &mut clients { if let ClientId::Client(client) = client && client_entity == *client { + debug!("writing message"); + debug!("lane: {:?}", message_registry.forward.get(M::type_path()).unwrap()); + debug!("registry: {:?}", message_registry.forward); transport.send.push( *message_registry.forward.get(M::type_path()) .expect("Failed to get message lane; the message likely isn't serialized yet"), to_allocvec(&message).expect("Failed to serialize message").into(), - Instant::now(), + bevy::platform::time::Instant::now(), ).expect("Failed to send message"); } } @@ -204,7 +265,26 @@ fn send_to_server( *message_registry.forward.get(M::type_path()) .expect("Failed to get message lane; the message likely isn't serialized yet"), to_allocvec(&message).expect("Failed to serialize message").into(), - Instant::now(), + bevy::platform::time::Instant::now(), + ).expect("Failed to send message"); + } + } +} +fn send_mapped_to_server( + mut messages: MessageReader, + mut sessions: Query<&mut Transport, Without>, + mut entity_map: ResMut, + message_registry: Res, +) { + for message in messages.read() { + for mut transport in &mut sessions { + let mut message = message.clone(); + message.map_entities(&mut entity_map.client_to_server); + transport.send.push( + *message_registry.forward.get(M::type_path()) + .expect("Failed to get message lane; the message likely isn't serialized yet"), + to_allocvec(&message).expect("Failed to serialize message").into(), + bevy::platform::time::Instant::now(), ).expect("Failed to send message"); } } @@ -220,6 +300,8 @@ fn recv_from_server( let mut messages = Vec::new(); for mut transport in sessions.iter_mut() { for message in transport.recv.msgs.drain() { + debug!("forward: {:?}", message_registry.forward); + debug!("reverse: {:?}", message_registry.reverse); let payload = message.payload; let message_fn = message_registry.reverse.get(&message.lane).expect("Packet was sent across a lane that didn't have a message assigned to it yet"); messages.push((*message_fn, payload));