Slightly better implementation of peers, still need to create a more generic system for deciding which components to distribute where and then use that for Peers.
This commit is contained in:
parent
e013fb427a
commit
53fe3333f0
14 changed files with 438 additions and 265 deletions
|
|
@ -2,13 +2,16 @@ use bevy::prelude::*;
|
|||
|
||||
use crate::net::prelude::*;
|
||||
|
||||
pub fn handle_new_peer(new_peers: Query<&Peer, Added<Peer>>) {
|
||||
pub fn handle_new_peer(new_peers: Query<&PeerID, Added<Peer>>) {
|
||||
for peer in new_peers {
|
||||
info!("Peer {} was added", peer.id);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn handle_deleted_peer(mut old_peers: RemovedComponents<Peer>, peers: Query<&Peer>) -> Result {
|
||||
pub fn handle_deleted_peer(
|
||||
mut old_peers: RemovedComponents<Peer>,
|
||||
peers: Query<&PeerID>,
|
||||
) -> Result {
|
||||
for entity in old_peers.read() {
|
||||
if let Ok(peer) = peers.get(entity) {
|
||||
info!("Peer {} was removed", peer.id);
|
||||
|
|
@ -20,7 +23,7 @@ pub fn handle_deleted_peer(mut old_peers: RemovedComponents<Peer>, peers: Query<
|
|||
}
|
||||
|
||||
pub fn handle_incoming_packets(mut packets: EventReader<InboundPacket>) {
|
||||
for packet in packets.read() {
|
||||
info!("Packet received: {:?}", packet.0.message);
|
||||
for InboundPacket(packet) in packets.read() {
|
||||
info!("Packet received: {:?}", packet.message);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,7 +13,10 @@ use super::{
|
|||
check_for_seed, setup_balls, setup_camera, setup_from_seed, setup_player, setup_walls,
|
||||
},
|
||||
state::AppState,
|
||||
ui::{setup_peer_ui, setup_seed_ui, update_peer_ui, update_peer_ui_timings, update_seed_ui},
|
||||
ui::{
|
||||
setup_peer_ui, setup_potential_peer_ui, setup_seed_ui, update_peer_ui,
|
||||
update_peer_ui_timings, update_potential_peer_ui, update_seed_ui,
|
||||
},
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
|
|
@ -55,11 +58,20 @@ impl GamePlugin {
|
|||
impl Plugin for GamePlugin {
|
||||
fn build(&self, app: &mut App) {
|
||||
app.add_plugins((
|
||||
NetIOPlugin::new(self.port, self.source.try_to_address()),
|
||||
NetIOPlugin::maybe_peer(self.port, self.source.try_to_address()),
|
||||
distribution_plugin::<Seed>,
|
||||
PhysicsPlugins::default().with_length_unit(50.0),
|
||||
))
|
||||
.init_state::<AppState>()
|
||||
.add_systems(Startup, (setup_camera, setup_seed_ui, setup_peer_ui))
|
||||
.add_systems(
|
||||
Startup,
|
||||
(
|
||||
setup_camera,
|
||||
setup_seed_ui,
|
||||
setup_peer_ui,
|
||||
setup_potential_peer_ui,
|
||||
),
|
||||
)
|
||||
.add_systems(
|
||||
OnEnter(AppState::InGame),
|
||||
(
|
||||
|
|
@ -75,8 +87,7 @@ impl Plugin for GamePlugin {
|
|||
handle_new_peer,
|
||||
handle_deleted_peer,
|
||||
handle_incoming_packets,
|
||||
)
|
||||
.run_if(in_state(NetworkState::MultiPlayer)),
|
||||
),
|
||||
),
|
||||
)
|
||||
.add_systems(
|
||||
|
|
@ -85,14 +96,15 @@ impl Plugin for GamePlugin {
|
|||
((move_player, move_camera).chain(), zoom_camera)
|
||||
.run_if(in_state(AppState::InGame)),
|
||||
update_seed_ui,
|
||||
(update_peer_ui, update_peer_ui_timings)
|
||||
.run_if(in_state(NetworkState::MultiPlayer)),
|
||||
(
|
||||
update_peer_ui,
|
||||
update_peer_ui_timings,
|
||||
update_potential_peer_ui,
|
||||
),
|
||||
quit.run_if(input_pressed(KeyCode::KeyQ)),
|
||||
),
|
||||
);
|
||||
|
||||
Seed::register(app);
|
||||
|
||||
match self.source {
|
||||
DataSource::Address(peer) => {
|
||||
info!("Will retrieve seed from peer => {peer}");
|
||||
|
|
|
|||
|
|
@ -61,7 +61,7 @@ impl From<Seed> for Vec<u8> {
|
|||
impl TryFrom<Vec<u8>> for Seed {
|
||||
type Error = TryFromSliceError;
|
||||
|
||||
fn try_from(value: Vec<u8>) -> Result<Self, Self::Error> {
|
||||
fn try_from(value: Vec<u8>) -> std::result::Result<Self, Self::Error> {
|
||||
Ok(TryInto::<[u8; 8]>::try_into(value.as_slice())?.into())
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use bevy::prelude::*;
|
||||
|
||||
use crate::net::prelude::{Peer, PeerReceiveTiming, PeerSendTiming};
|
||||
use crate::net::prelude::{Peer, PeerID, PeerReceiveTiming, PeerSendTiming, PotentialPeers};
|
||||
|
||||
use super::seed::Seed;
|
||||
|
||||
|
|
@ -54,11 +54,11 @@ pub fn setup_peer_ui(mut commands: Commands) {
|
|||
}
|
||||
|
||||
#[derive(Component)]
|
||||
pub struct PeerID(Entity);
|
||||
pub struct UIPeerID(Entity);
|
||||
|
||||
pub fn update_peer_ui(
|
||||
ui: Query<(Entity, &Children), With<PeerUI>>,
|
||||
rows: Query<&PeerID>,
|
||||
rows: Query<&UIPeerID>,
|
||||
added: Query<Entity, Added<Peer>>,
|
||||
mut removed: RemovedComponents<Peer>,
|
||||
mut commands: Commands,
|
||||
|
|
@ -67,14 +67,14 @@ pub fn update_peer_ui(
|
|||
for addition in added {
|
||||
commands
|
||||
.entity(table)
|
||||
.with_child((Text::new("---- ---- ----"), PeerID(addition)));
|
||||
.with_child((Text::new("---- ---- ----"), UIPeerID(addition)));
|
||||
}
|
||||
for removal in removed.read() {
|
||||
for child in children {
|
||||
if let Ok(id) = rows.get(*child) {
|
||||
if id.0 == removal {
|
||||
commands.entity(*child).despawn();
|
||||
}
|
||||
if let Ok(&UIPeerID(id)) = rows.get(*child)
|
||||
&& id == removal
|
||||
{
|
||||
commands.entity(*child).despawn();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -82,12 +82,12 @@ pub fn update_peer_ui(
|
|||
}
|
||||
|
||||
pub fn update_peer_ui_timings(
|
||||
rows: Query<(&mut Text, &PeerID)>,
|
||||
peers: Query<(&Peer, &PeerReceiveTiming, &PeerSendTiming)>,
|
||||
rows: Query<(&mut Text, &UIPeerID)>,
|
||||
peers: Query<(&PeerID, &PeerReceiveTiming, &PeerSendTiming)>,
|
||||
time: Res<Time>,
|
||||
) {
|
||||
for (mut row, id) in rows {
|
||||
if let Ok((peer, recv, send)) = peers.get(id.0) {
|
||||
for (mut row, &UIPeerID(id)) in rows {
|
||||
if let Ok((peer, recv, send)) = peers.get(id) {
|
||||
**row = format!(
|
||||
"{} {:.2} {:.2}",
|
||||
peer.id,
|
||||
|
|
@ -97,3 +97,34 @@ pub fn update_peer_ui_timings(
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Component, Debug)]
|
||||
pub struct PotentialPeerUI;
|
||||
|
||||
pub fn setup_potential_peer_ui(mut commands: Commands) {
|
||||
commands
|
||||
.spawn((
|
||||
Text::new("Potential peers:"),
|
||||
Node {
|
||||
position_type: PositionType::Absolute,
|
||||
bottom: Val::Px(5.0),
|
||||
right: Val::Px(5.0),
|
||||
..default()
|
||||
},
|
||||
))
|
||||
.with_child((TextSpan::new(""), PotentialPeerUI));
|
||||
}
|
||||
|
||||
pub fn update_potential_peer_ui(
|
||||
potential_peers: Res<PotentialPeers>,
|
||||
text: Query<&mut TextSpan, With<PotentialPeerUI>>,
|
||||
) {
|
||||
for mut span in text {
|
||||
**span = potential_peers
|
||||
.addresses
|
||||
.iter()
|
||||
.map(|addr| addr.to_string())
|
||||
.collect::<Vec<String>>()
|
||||
.join("\n");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ use game::prelude::*;
|
|||
/// Also functions as a Bevy plugin to pass the configuration into the app.
|
||||
#[derive(Parser)]
|
||||
#[command(version, about)]
|
||||
pub struct AppSettings {
|
||||
pub struct AppSettingsPlugin {
|
||||
#[command(flatten)]
|
||||
source: Source,
|
||||
|
||||
|
|
@ -36,7 +36,7 @@ struct Source {
|
|||
connect: Option<SocketAddr>,
|
||||
}
|
||||
|
||||
impl Plugin for AppSettings {
|
||||
impl Plugin for AppSettingsPlugin {
|
||||
fn build(&self, app: &mut App) {
|
||||
app.insert_resource(Gravity(Vector::ZERO)).add_plugins((
|
||||
DefaultPlugins.set(WindowPlugin {
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
use bevy::prelude::{App, AppExit};
|
||||
use clap::Parser;
|
||||
|
||||
use distributed_physics_test::AppSettings;
|
||||
use distributed_physics_test::AppSettingsPlugin;
|
||||
|
||||
fn main() -> AppExit {
|
||||
App::new().add_plugins(AppSettings::parse()).run()
|
||||
App::new().add_plugins(AppSettingsPlugin::parse()).run()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,48 +2,32 @@ use bevy::prelude::*;
|
|||
|
||||
use super::{
|
||||
packet::{InboundPacket, OutboundPacket, Packet},
|
||||
peer::Peer,
|
||||
state::NetworkState,
|
||||
peer::{Peer, PeerID},
|
||||
};
|
||||
|
||||
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()) {
|
||||
for InboundPacket(packet) in inbound.read() {
|
||||
if let Ok(entity) = T::try_from(packet.message.clone()) {
|
||||
commands.spawn(entity);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn sender<T: Into<Vec<u8>> + Component + Clone>(
|
||||
peers: Query<&Peer, Added<Peer>>,
|
||||
peers: Query<&PeerID, 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,
|
||||
)));
|
||||
outbound.write(Packet::create((*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)),
|
||||
);
|
||||
}
|
||||
pub fn distribution_plugin<T: Into<Vec<u8>> + TryFrom<Vec<u8>> + Component + Clone>(app: &mut App) {
|
||||
app.add_systems(FixedUpdate, (sender::<T>, spawner::<T>));
|
||||
}
|
||||
|
|
|
|||
77
src/net/heartbeat.rs
Normal file
77
src/net/heartbeat.rs
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
use std::time::Duration;
|
||||
|
||||
use bevy::prelude::*;
|
||||
|
||||
use crate::net::{packet::PacketType, peer::PotentialPeers};
|
||||
|
||||
use super::{
|
||||
io::{Config, format_message},
|
||||
packet::{OutboundPacket, Packet},
|
||||
peer::{Peer, PeerChangeEvent, PeerID, PeerReceiveTiming, PeerSendTiming},
|
||||
queues::NetworkSend,
|
||||
};
|
||||
|
||||
const PING_FREQUENCY: Duration = Duration::from_secs(3);
|
||||
const MISSED_PINGS: u32 = 3;
|
||||
|
||||
// TODO: Perhaps this needs a state rethink, is Single/Multiplayer actually useful vs Disconnected, Connecting, Connected?
|
||||
// Would also help to state-scope some of these things, like InitialAddresses vs PeerMap
|
||||
pub fn heartbeat(
|
||||
peers: Query<(&PeerID, &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() + PING_FREQUENCY > time.elapsed() {
|
||||
continue;
|
||||
}
|
||||
outbound.write(Packet::create(Vec::new(), peer.id));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn timeout(
|
||||
peers: Query<(&PeerID, &PeerReceiveTiming), With<Peer>>, // I mean... With<Peer> is inherent, but I guess I'll keep it as that might change
|
||||
time: Res<Time>,
|
||||
mut delete: EventWriter<PeerChangeEvent>,
|
||||
) {
|
||||
for (peer, last) in peers {
|
||||
if last.time() + PING_FREQUENCY * MISSED_PINGS < time.elapsed() {
|
||||
warn!("Peer {} timed out", peer.id);
|
||||
delete.write(PeerChangeEvent::new(peer.id, None));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Resource)]
|
||||
pub struct PotentialPeerTimer {
|
||||
timer: Timer,
|
||||
}
|
||||
|
||||
impl Default for PotentialPeerTimer {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
timer: Timer::new(PING_FREQUENCY, TimerMode::Repeating),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn ping_potential_peers(
|
||||
mut timer: ResMut<PotentialPeerTimer>,
|
||||
time: Res<Time>,
|
||||
peers: Res<PotentialPeers>,
|
||||
to_socket: Res<NetworkSend>,
|
||||
config: Res<Config>,
|
||||
) -> Result {
|
||||
timer.timer.tick(time.delta());
|
||||
if timer.timer.finished() {
|
||||
for peer in &peers.addresses {
|
||||
to_socket.send(
|
||||
format_message(&Vec::new(), PacketType::Peer, config.id),
|
||||
*peer,
|
||||
)?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -1,16 +1,29 @@
|
|||
use std::time::Duration;
|
||||
|
||||
use bevy::prelude::*;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::net::peer::PotentialPeer;
|
||||
use crate::net::packet::PacketType;
|
||||
|
||||
use super::{
|
||||
packet::{InboundPacket, OutboundPacket, Packet},
|
||||
peer::{Peer, PeerChangeEvent, PeerData, PeerMap, PeerReceiveTiming, PeerSendTiming},
|
||||
peer::{PeerChangeEvent, PeerData, PeerMap, PeerReceiveTiming, PeerSendTiming},
|
||||
queues::{NetworkReceive, NetworkSend},
|
||||
};
|
||||
|
||||
#[derive(Debug, Resource)]
|
||||
pub struct Config {
|
||||
pub id: Uuid,
|
||||
}
|
||||
|
||||
impl Default for Config {
|
||||
fn default() -> Self {
|
||||
Self { id: Uuid::new_v4() }
|
||||
}
|
||||
}
|
||||
|
||||
pub fn format_message(data: &Vec<u8>, variant: PacketType, id: Uuid) -> Vec<u8> {
|
||||
[data.as_slice(), &[variant as u8], id.as_bytes()].concat()
|
||||
}
|
||||
|
||||
pub fn handle_network_input(
|
||||
from_socket: Res<NetworkReceive>,
|
||||
peer_map: Res<PeerMap>,
|
||||
|
|
@ -19,26 +32,24 @@ pub fn handle_network_input(
|
|||
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 {} is not large enough to contain UUID",
|
||||
message.len()
|
||||
)
|
||||
.into());
|
||||
}
|
||||
let uuid = Uuid::from_slice(message.split_off(message.len() - 16).as_slice())?;
|
||||
if !message.is_empty() {
|
||||
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;
|
||||
for (message, address) in from_socket.iter() {
|
||||
match Packet::try_from(message) {
|
||||
Ok(packet) => {
|
||||
// TODO: Handle packet variant
|
||||
if !packet.message.is_empty() {
|
||||
to_app.write(packet.clone().into());
|
||||
}
|
||||
if let Some(peer_id) = peer_map.get(&packet.peer) {
|
||||
let (peer, mut last) = peers.get_mut(*peer_id)?;
|
||||
last.update(&time);
|
||||
if address == peer.addr {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
change_peer.write(PeerChangeEvent::new(packet.peer, Some(address)));
|
||||
}
|
||||
Err(err) => warn!("Error reading packet: {:?}", err),
|
||||
}
|
||||
change_peer.write(PeerChangeEvent::new(uuid, Some(address)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -47,52 +58,16 @@ pub fn handle_network_output(
|
|||
mut from_app: EventReader<OutboundPacket>,
|
||||
peer_map: Res<PeerMap>,
|
||||
mut peers: Query<(&PeerData, &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)?;
|
||||
for OutboundPacket(packet) in from_app.read() {
|
||||
let peer_id = peer_map.try_get(&packet.peer)?;
|
||||
let (peer, mut last) = peers.get_mut(*peer_id)?;
|
||||
// Append our UUID for client identification
|
||||
let message = [packet.0.message.as_slice(), peer.me.as_bytes()].concat();
|
||||
let message = format_message(&packet.message, packet.variant, config.id);
|
||||
to_socket.send(message, peer.addr.into())?;
|
||||
last.update(&time);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
const TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
pub fn heartbeat(
|
||||
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;
|
||||
}
|
||||
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(
|
||||
peers: Query<(&Peer, &PeerReceiveTiming)>,
|
||||
time: Res<Time>,
|
||||
mut delete: EventWriter<PeerChangeEvent>,
|
||||
) {
|
||||
for (peer, last) in peers {
|
||||
if last.time() + TIMEOUT < time.elapsed() {
|
||||
warn!("Peer {} timed out", peer.id);
|
||||
delete.write(PeerChangeEvent::new(peer.id, None));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,18 +1,17 @@
|
|||
mod distribution;
|
||||
mod heartbeat;
|
||||
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::distribution::distribution_plugin;
|
||||
pub use super::packet::{InboundPacket, OutboundPacket, Packet};
|
||||
pub use super::peer::{Peer, PeerReceiveTiming, PeerSendTiming};
|
||||
pub use super::peer::{Peer, PeerID, PeerReceiveTiming, PeerSendTiming, PotentialPeers};
|
||||
pub use super::plugin::NetIOPlugin;
|
||||
pub use super::state::NetworkState;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,19 +2,95 @@ use bevy::prelude::*;
|
|||
use uuid::Uuid;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum TryFromBytesError {
|
||||
InsufficientLength,
|
||||
NotUUID,
|
||||
NotVariant,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
pub enum PacketType {
|
||||
Standard = 0x00,
|
||||
Peer = 0x01,
|
||||
}
|
||||
|
||||
impl From<PacketType> for u8 {
|
||||
fn from(value: PacketType) -> Self {
|
||||
value as u8
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<u8> for PacketType {
|
||||
type Error = TryFromBytesError;
|
||||
|
||||
fn try_from(value: u8) -> std::result::Result<Self, Self::Error> {
|
||||
match value {
|
||||
value if value == PacketType::Standard as u8 => Ok(PacketType::Standard),
|
||||
value if value == PacketType::Peer as u8 => Ok(PacketType::Peer),
|
||||
_ => Err(TryFromBytesError::NotVariant),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub const UUID_SIZE: usize = 16;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Packet {
|
||||
pub message: Vec<u8>,
|
||||
pub variant: PacketType,
|
||||
pub peer: Uuid,
|
||||
}
|
||||
|
||||
impl Packet {
|
||||
pub fn new(message: Vec<u8>, peer: Uuid) -> Self {
|
||||
Self { peer, message }
|
||||
pub fn new(message: Vec<u8>, variant: PacketType, peer: Uuid) -> Self {
|
||||
Self {
|
||||
message,
|
||||
variant,
|
||||
peer,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn create<T: From<Packet>>(message: Vec<u8>, peer: Uuid) -> T {
|
||||
Self {
|
||||
message,
|
||||
variant: PacketType::Standard,
|
||||
peer,
|
||||
}
|
||||
.into()
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<Vec<u8>> for Packet {
|
||||
type Error = TryFromBytesError;
|
||||
|
||||
fn try_from(mut value: Vec<u8>) -> std::result::Result<Self, Self::Error> {
|
||||
if value.len() < UUID_SIZE {
|
||||
return Err(TryFromBytesError::InsufficientLength);
|
||||
}
|
||||
let uuid = Uuid::from_slice(value.split_off(value.len() - UUID_SIZE).as_slice())
|
||||
.map_err(|_| TryFromBytesError::NotUUID)?;
|
||||
let variant = value
|
||||
.pop()
|
||||
.ok_or(TryFromBytesError::InsufficientLength)?
|
||||
.try_into()?;
|
||||
Ok(Packet::new(value, variant, uuid))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Event)]
|
||||
pub struct OutboundPacket(pub Packet);
|
||||
|
||||
impl From<Packet> for OutboundPacket {
|
||||
fn from(value: Packet) -> Self {
|
||||
Self(value)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Event)]
|
||||
pub struct InboundPacket(pub Packet);
|
||||
|
||||
impl From<Packet> for InboundPacket {
|
||||
fn from(value: Packet) -> Self {
|
||||
Self(value)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
226
src/net/peer.rs
226
src/net/peer.rs
|
|
@ -1,14 +1,14 @@
|
|||
use std::{
|
||||
array::TryFromSliceError,
|
||||
collections::HashMap,
|
||||
net::{IpAddr, Ipv6Addr, SocketAddr},
|
||||
collections::{HashMap, HashSet},
|
||||
net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use bevy::prelude::*;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::packet::{OutboundPacket, Packet};
|
||||
use super::packet::{InboundPacket, OutboundPacket, Packet, PacketType};
|
||||
|
||||
#[derive(Component, Debug, Default)]
|
||||
pub struct PeerSendTiming(Duration);
|
||||
|
|
@ -40,55 +40,12 @@ impl PeerReceiveTiming {
|
|||
}
|
||||
}
|
||||
|
||||
#[derive(Component, Debug)]
|
||||
#[require(PeerData, PeerReceiveTiming)]
|
||||
pub struct Peer {
|
||||
pub id: 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
|
||||
fn from(Address(value): Address) -> Self {
|
||||
value
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -111,12 +68,12 @@ impl PartialEq<Address> for SocketAddr {
|
|||
}
|
||||
|
||||
impl From<Address> for Vec<u8> {
|
||||
fn from(value: Address) -> Self {
|
||||
let mut bytes: Vec<u8> = match value.0.ip() {
|
||||
fn from(Address(value): Address) -> Self {
|
||||
let mut bytes: Vec<u8> = match value.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.extend(value.port().to_le_bytes());
|
||||
bytes
|
||||
}
|
||||
}
|
||||
|
|
@ -124,14 +81,25 @@ impl From<Address> for Vec<u8> {
|
|||
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(),
|
||||
fn try_from(value: Vec<u8>) -> std::result::Result<Self, Self::Error> {
|
||||
const PORT_SIZE: usize = 2;
|
||||
|
||||
if value.len() < PORT_SIZE {
|
||||
todo!();
|
||||
}
|
||||
let mut bytes = value.clone();
|
||||
let port = u16::from_le_bytes(TryInto::<[u8; PORT_SIZE]>::try_into(
|
||||
bytes.split_off(bytes.len() - PORT_SIZE).as_slice(),
|
||||
)?);
|
||||
let addr = if let Ok(bytes) = TryInto::<[u8; 4]>::try_into(value.as_slice()) {
|
||||
let addr = if let Ok(bytes) =
|
||||
TryInto::<[u8; Ipv4Addr::BITS as usize / 8]>::try_into(bytes.as_slice())
|
||||
{
|
||||
SocketAddr::from((bytes, port))
|
||||
} else {
|
||||
SocketAddr::from((TryInto::<[u8; 16]>::try_into(value.as_slice())?, port))
|
||||
SocketAddr::from((
|
||||
TryInto::<[u8; Ipv6Addr::BITS as usize / 8]>::try_into(bytes.as_slice())?,
|
||||
port,
|
||||
))
|
||||
};
|
||||
Ok(Address(addr))
|
||||
}
|
||||
|
|
@ -143,6 +111,53 @@ impl Default for Address {
|
|||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Component, Copy, Debug, Default)]
|
||||
#[require(PeerSendTiming)]
|
||||
pub struct PeerData {
|
||||
pub addr: Address,
|
||||
}
|
||||
|
||||
impl PeerData {
|
||||
pub fn new(addr: SocketAddr) -> Self {
|
||||
Self { addr: addr.into() }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Component, Debug)]
|
||||
#[require(PeerData)]
|
||||
pub struct PeerID {
|
||||
pub id: Uuid,
|
||||
}
|
||||
|
||||
impl PeerID {
|
||||
pub fn new(id: Uuid) -> Self {
|
||||
Self { id }
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for PeerID {
|
||||
fn default() -> Self {
|
||||
Self { id: Uuid::new_v4() }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Component, Debug)]
|
||||
#[require(PeerID, PeerReceiveTiming)]
|
||||
pub struct Peer;
|
||||
|
||||
#[derive(Debug, Default, Resource)]
|
||||
pub struct PotentialPeers {
|
||||
pub addresses: HashSet<SocketAddr>,
|
||||
}
|
||||
|
||||
impl PotentialPeers {
|
||||
pub fn new(addresses: Vec<SocketAddr>) -> Self {
|
||||
Self {
|
||||
addresses: addresses.into_iter().collect(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Resource)]
|
||||
pub struct PeerMap {
|
||||
map: HashMap<Uuid, Entity>,
|
||||
|
|
@ -157,17 +172,13 @@ impl PeerMap {
|
|||
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 try_get(&self, uuid: &Uuid) -> std::result::Result<&Entity, String> {
|
||||
self.get(uuid).ok_or(format!("No peer 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)]
|
||||
|
|
@ -186,63 +197,82 @@ pub fn handle_peer_change(
|
|||
mut changes: EventReader<PeerChangeEvent>,
|
||||
mut peer_map: ResMut<PeerMap>,
|
||||
mut peers: Query<&mut PeerData>,
|
||||
mut potential_peers: ResMut<PotentialPeers>,
|
||||
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 {
|
||||
match (peer_map.get(&change.peer), change.addr) {
|
||||
// Peer modification
|
||||
(Some(entity), Some(addr)) => {
|
||||
if let Ok(mut peer) = peers.get_mut(*entity) {
|
||||
peer.addr = Address(addr);
|
||||
peer.addr = addr.into();
|
||||
} else {
|
||||
warn!("Peer {} doesn't exist (just added?)", change.peer);
|
||||
}
|
||||
} else {
|
||||
}
|
||||
// Peer deletion
|
||||
(Some(entity), None) => {
|
||||
commands.get_entity(*entity)?.despawn();
|
||||
peer_map.remove(&change.peer);
|
||||
}
|
||||
} else if let Some(addr) = change.addr {
|
||||
info!("Adding peer {} ({})", change.peer, addr);
|
||||
peer_map.insert(
|
||||
change.peer,
|
||||
commands
|
||||
.spawn((
|
||||
PeerData::new(addr, Uuid::new_v4()),
|
||||
PeerReceiveTiming::new(&time),
|
||||
))
|
||||
.id(),
|
||||
);
|
||||
} else {
|
||||
warn!("Peer {} already deleted", change.peer);
|
||||
// Peer addition
|
||||
(None, Some(addr)) => {
|
||||
info!("Adding peer {} ({})", change.peer, addr);
|
||||
peer_map.insert(
|
||||
change.peer,
|
||||
commands
|
||||
.spawn((
|
||||
Peer,
|
||||
PeerID::new(change.peer),
|
||||
PeerData::new(addr),
|
||||
PeerReceiveTiming::new(&time),
|
||||
))
|
||||
.id(),
|
||||
);
|
||||
}
|
||||
// Double peer deletion
|
||||
(None, None) => warn!("Peer {} already deleted", change.peer),
|
||||
}
|
||||
}
|
||||
if peer_map.len() > 1 {
|
||||
if let Some(entity) = peer_map.remove(&Uuid::nil()) {
|
||||
commands.get_entity(entity)?.despawn();
|
||||
if let Some(addr) = change.addr {
|
||||
potential_peers.addresses.remove(&addr);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[allow(clippy::type_complexity)]
|
||||
pub fn new_peer_message(
|
||||
mut from_network: EventReader<InboundPacket>,
|
||||
peers: Query<&PeerData>,
|
||||
mut potential_peers: ResMut<PotentialPeers>,
|
||||
) {
|
||||
'packet: for packet in from_network.read() {
|
||||
if let Ok(addr) = TryInto::<Address>::try_into(packet.0.message.clone()) {
|
||||
if potential_peers.addresses.contains(&addr.into()) {
|
||||
continue;
|
||||
}
|
||||
for peer in peers {
|
||||
if peer.addr == addr.into() {
|
||||
continue 'packet;
|
||||
}
|
||||
}
|
||||
potential_peers.addresses.insert(addr.into());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: Make this a more generic system
|
||||
pub fn handle_new_peer(
|
||||
new_peers: Query<(Option<Ref<Peer>>, Option<&Peer>, &PeerData)>,
|
||||
peers: Query<(Ref<Peer>, &PeerID, &PeerData)>,
|
||||
mut outbound: EventWriter<OutboundPacket>,
|
||||
) -> 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());
|
||||
) {
|
||||
for (change, peer, _) in peers {
|
||||
if change.is_added() {
|
||||
for (_, other, data) in peers {
|
||||
if peer.id != other.id {
|
||||
outbound.write(Packet::new(data.addr.into(), PacketType::Peer, peer.id).into());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,65 +1,59 @@
|
|||
use std::net::SocketAddr;
|
||||
|
||||
use bevy::prelude::*;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{
|
||||
io::{handle_network_input, handle_network_output, heartbeat, timeout},
|
||||
heartbeat::{PotentialPeerTimer, heartbeat, ping_potential_peers, timeout},
|
||||
io::{Config, handle_network_input, handle_network_output},
|
||||
packet::{InboundPacket, OutboundPacket},
|
||||
peer::{PeerChangeEvent, PeerData, PeerMap, handle_new_peer, handle_peer_change},
|
||||
peer::{
|
||||
PeerChangeEvent, PeerMap, PotentialPeers, handle_new_peer, handle_peer_change,
|
||||
new_peer_message,
|
||||
},
|
||||
queues::{NetworkReceive, NetworkSend},
|
||||
socket::bind_socket,
|
||||
state::NetworkState,
|
||||
};
|
||||
|
||||
pub struct NetIOPlugin {
|
||||
listen: u16,
|
||||
peer: Option<SocketAddr>,
|
||||
initial_peers: Vec<SocketAddr>,
|
||||
}
|
||||
|
||||
impl NetIOPlugin {
|
||||
pub fn new(listen: u16, peer: Option<SocketAddr>) -> Self {
|
||||
Self { listen, peer }
|
||||
pub fn maybe_peer(listen: u16, peer: Option<SocketAddr>) -> Self {
|
||||
Self {
|
||||
listen,
|
||||
initial_peers: match peer {
|
||||
Some(addr) => vec![addr],
|
||||
None => Vec::new(),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Plugin for NetIOPlugin {
|
||||
fn build(&self, app: &mut App) {
|
||||
app.init_state::<NetworkState>()
|
||||
.add_systems(
|
||||
FixedPreUpdate,
|
||||
(handle_network_input, handle_peer_change)
|
||||
.chain()
|
||||
.run_if(in_state(NetworkState::MultiPlayer)),
|
||||
)
|
||||
.add_systems(
|
||||
FixedUpdate,
|
||||
(heartbeat, timeout, handle_new_peer).run_if(in_state(NetworkState::MultiPlayer)),
|
||||
)
|
||||
.add_systems(
|
||||
FixedPostUpdate,
|
||||
handle_network_output.run_if(in_state(NetworkState::MultiPlayer)),
|
||||
)
|
||||
.add_event::<PeerChangeEvent>()
|
||||
.add_event::<InboundPacket>()
|
||||
.add_event::<OutboundPacket>();
|
||||
app.add_systems(
|
||||
FixedPreUpdate,
|
||||
(handle_network_input, (handle_peer_change, new_peer_message)).chain(),
|
||||
)
|
||||
.add_systems(
|
||||
FixedUpdate,
|
||||
(heartbeat, timeout, handle_new_peer, ping_potential_peers),
|
||||
)
|
||||
.add_systems(FixedPostUpdate, handle_network_output)
|
||||
.init_resource::<Config>()
|
||||
.init_resource::<PeerMap>()
|
||||
.init_resource::<PotentialPeerTimer>()
|
||||
.insert_resource(PotentialPeers::new(self.initial_peers.clone()))
|
||||
.add_event::<PeerChangeEvent>()
|
||||
.add_event::<InboundPacket>()
|
||||
.add_event::<OutboundPacket>();
|
||||
|
||||
match bind_socket(self.listen) {
|
||||
Ok((send, receive)) => {
|
||||
app.insert_state(NetworkState::MultiPlayer)
|
||||
.insert_resource(NetworkSend::new(send))
|
||||
app.insert_resource(NetworkSend::new(send))
|
||||
.insert_resource(NetworkReceive::new(receive));
|
||||
|
||||
let mut peer_map = PeerMap::default();
|
||||
if let Some(socket) = self.peer {
|
||||
let entity = app.world_mut().spawn(PeerData {
|
||||
addr: socket.into(),
|
||||
me: Uuid::nil(),
|
||||
});
|
||||
peer_map.insert(Uuid::nil(), entity.id());
|
||||
}
|
||||
|
||||
app.insert_resource(peer_map);
|
||||
}
|
||||
Err(err) => {
|
||||
warn!("Failed to set up networking: {err}");
|
||||
|
|
|
|||
|
|
@ -1,8 +0,0 @@
|
|||
use bevy::prelude::*;
|
||||
|
||||
#[derive(States, Default, Debug, Clone, PartialEq, Eq, Hash)]
|
||||
pub enum NetworkState {
|
||||
#[default]
|
||||
SinglePlayer,
|
||||
MultiPlayer,
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue