Mirrored_Repos/veloren
1
0
Fork 0
mirror of https://gitlab.com/veloren/veloren.git synced 2024-08-30 18:12:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

remote all streams from Client and put it directly in the ecs system.

Browse Source 
This commit does not run compile as the message sys now would requiere 30 imput parameters which is over the max of 26 :/
This commit is contained in:
Marcel Märtens 2020-10-16 21:37:28 +02:00
parent 0e32cedd85
commit dd966dd00e
17 changed files with 707 additions and 461 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
server/src/client.rs
View File

@ -1,24 +1,39 @@
use crate::error::Error; use common::msg::{ClientInGame, ClientType};
use common::msg::{ClientInGame, ClientType, ServerGeneral, ServerMsg};
use hashbrown::HashSet; use hashbrown::HashSet;
use network::{Participant, Stream}; use network::{Participant, Stream};
use serde::{de::DeserializeOwned, Serialize};
use specs::{Component, FlaggedStorage}; use specs::{Component, FlaggedStorage};
use specs_idvs::IdvStorage; use specs_idvs::IdvStorage;
use tracing::debug;
use vek::*; use vek::*;
// Streams
// we ignore errors on send, and do unified error handling in recv
pub struct GeneralStream(pub Stream);
pub struct PingStream(pub Stream);
pub struct RegisterStream(pub Stream);
pub struct CharacterScreenStream(pub Stream);
pub struct InGameStream(pub Stream);
impl Component for GeneralStream {
type Storage = FlaggedStorage<Self, IdvStorage<Self>>;
}
impl Component for PingStream {
type Storage = FlaggedStorage<Self, IdvStorage<Self>>;
}
impl Component for RegisterStream {
type Storage = FlaggedStorage<Self, IdvStorage<Self>>;
}
impl Component for CharacterScreenStream {
type Storage = FlaggedStorage<Self, IdvStorage<Self>>;
}
impl Component for InGameStream {
type Storage = FlaggedStorage<Self, IdvStorage<Self>>;
}
pub struct Client { pub struct Client {
pub registered: bool, pub registered: bool,
pub client_type: ClientType, pub client_type: ClientType,
pub in_game: Option<ClientInGame>, pub in_game: Option<ClientInGame>,
pub participant: Option<Participant>, pub participant: Option<Participant>,
pub general_stream: Stream,
pub ping_stream: Stream,
pub register_stream: Stream,
pub character_screen_stream: Stream,
pub in_game_stream: Stream,
pub network_error: bool,
pub last_ping: f64, pub last_ping: f64,
pub login_msg_sent: bool, pub login_msg_sent: bool,
} }
@ -27,97 +42,6 @@ impl Component for Client {
type Storage = FlaggedStorage<Self, IdvStorage<Self>>; type Storage = FlaggedStorage<Self, IdvStorage<Self>>;
} }
impl Client {
fn internal_send<M: Serialize>(err: &mut bool, s: &mut Stream, msg: M) {
if !*err {
if let Err(e) = s.send(msg) {
debug!(?e, "got a network error with client");
*err = true;
}
}
}
/*
fn internal_send_raw(b: &AtomicBool, s: &mut Stream, msg: Arc<MessageBuffer>) {
if !b.load(Ordering::Relaxed) {
if let Err(e) = s.send_raw(msg) {
debug!(?e, "got a network error with client");
b.store(true, Ordering::Relaxed);
}
}
}
*/
pub fn send_msg<S>(&mut self, msg: S)
where
S: Into<ServerMsg>,
{
const ERR: &str =
"Don't do that. Sending these messages is only done ONCE at connect and not by this fn";
match msg.into() {
ServerMsg::Info(_) => panic!(ERR),
ServerMsg::Init(_) => panic!(ERR),
ServerMsg::RegisterAnswer(msg) => {
Self::internal_send(&mut self.network_error, &mut self.register_stream, &msg)
},
ServerMsg::General(msg) => {
let stream = match &msg {
//Character Screen related
ServerGeneral::CharacterDataLoadError(_)
| ServerGeneral::CharacterListUpdate(_)
| ServerGeneral::CharacterActionError(_)
| ServerGeneral::CharacterSuccess => &mut self.character_screen_stream,
//Ingame related
ServerGeneral::GroupUpdate(_)
| ServerGeneral::GroupInvite { .. }
| ServerGeneral::InvitePending(_)
| ServerGeneral::InviteComplete { .. }
| ServerGeneral::ExitInGameSuccess
| ServerGeneral::InventoryUpdate(_, _)
| ServerGeneral::TerrainChunkUpdate { .. }
| ServerGeneral::TerrainBlockUpdates(_)
| ServerGeneral::SetViewDistance(_)
| ServerGeneral::Outcomes(_)
| ServerGeneral::Knockback(_) => &mut self.in_game_stream,
// Always possible
ServerGeneral::PlayerListUpdate(_)
| ServerGeneral::ChatMsg(_)
| ServerGeneral::SetPlayerEntity(_)
| ServerGeneral::TimeOfDay(_)
| ServerGeneral::EntitySync(_)
| ServerGeneral::CompSync(_)
| ServerGeneral::CreateEntity(_)
| ServerGeneral::DeleteEntity(_)
| ServerGeneral::Disconnect(_)
| ServerGeneral::Notification(_) => &mut self.general_stream,
};
Self::internal_send(&mut self.network_error, stream, &msg)
},
ServerMsg::Ping(msg) => {
Self::internal_send(&mut self.network_error, &mut self.ping_stream, &msg)
},
};
}
pub async fn internal_recv<M: DeserializeOwned>(
err: &mut bool,
s: &mut Stream,
) -> Result<M, Error> {
if !*err {
match s.recv().await {
Ok(r) => Ok(r),
Err(e) => {
debug!(?e, "got a network error with client while recv");
*err = true;
Err(Error::StreamErr(e))
},
}
} else {
Err(Error::StreamErr(network::StreamError::StreamClosed))
}
}
}
// Distance from fuzzy_chunk before snapping to current chunk // Distance from fuzzy_chunk before snapping to current chunk
pub const CHUNK_FUZZ: u32 = 2; pub const CHUNK_FUZZ: u32 = 2;
// Distance out of the range of a region before removing it from subscriptions // Distance out of the range of a region before removing it from subscriptions

16
server/src/cmd.rs
View File

@ -3,7 +3,6 @@
//! `CHAT_COMMANDS` and provide a handler function. //! `CHAT_COMMANDS` and provide a handler function.
use crate::{ use crate::{
client::Client,
settings::{BanRecord, EditableSetting}, settings::{BanRecord, EditableSetting},
Server, StateExt, Server, StateExt,
}; };
@ -27,7 +26,7 @@ use std::convert::TryFrom;
use vek::*; use vek::*;
use world::util::Sampler; use world::util::Sampler;
use crate::login_provider::LoginProvider; use crate::{client::InGameStream, login_provider::LoginProvider};
use scan_fmt::{scan_fmt, scan_fmt_some}; use scan_fmt::{scan_fmt, scan_fmt_some};
use tracing::error; use tracing::error;
@ -650,7 +649,8 @@ fn handle_spawn(
// Add to group system if a pet // Add to group system if a pet
if matches!(alignment, comp::Alignment::Owned { .. }) { if matches!(alignment, comp::Alignment::Owned { .. }) {
let state = server.state(); let state = server.state();
let mut clients = state.ecs().write_storage::<Client>(); let mut in_game_streams =
state.ecs().write_storage::<InGameStream>();
let uids = state.ecs().read_storage::<Uid>(); let uids = state.ecs().read_storage::<Uid>();
let mut group_manager = let mut group_manager =
state.ecs().write_resource::<comp::group::GroupManager>(); state.ecs().write_resource::<comp::group::GroupManager>();
@ -662,15 +662,15 @@ fn handle_spawn(
&state.ecs().read_storage(), &state.ecs().read_storage(),
&uids, &uids,
&mut |entity, group_change| { &mut |entity, group_change| {
clients in_game_streams
.get_mut(entity) .get_mut(entity)
.and_then(|c| { .and_then(|s| {
group_change group_change
.try_map(|e| uids.get(e).copied()) .try_map(|e| uids.get(e).copied())
.map(|g| (g, c)) .map(|g| (g, s))
}) })
.map(|(g, c)| { .map(|(g, s)| {
c.send_msg(ServerGeneral::GroupUpdate(g)) let _ = s.0.send(ServerGeneral::GroupUpdate(g));
}); });
}, },
); );

39
server/src/connection_handler.rs
View File

@ -1,4 +1,7 @@
use crate::{Client, ClientType, ServerInfo}; use crate::{
CharacterScreenStream, Client, ClientType, GeneralStream, InGameStream, PingStream,
RegisterStream, ServerInfo,
};
use crossbeam::{bounded, unbounded, Receiver, Sender}; use crossbeam::{bounded, unbounded, Receiver, Sender};
use futures_channel::oneshot; use futures_channel::oneshot;
use futures_executor::block_on; use futures_executor::block_on;
@ -13,10 +16,19 @@ pub(crate) struct ServerInfoPacket {
pub time: f64, pub time: f64,
} }
pub(crate) struct ClientPackage {
pub client: Client,
pub general: GeneralStream,
pub ping: PingStream,
pub register: RegisterStream,
pub character: CharacterScreenStream,
pub in_game: InGameStream,
}
pub(crate) struct ConnectionHandler { pub(crate) struct ConnectionHandler {
_network: Arc<Network>, _network: Arc<Network>,
thread_handle: Option<thread::JoinHandle<()>>, thread_handle: Option<thread::JoinHandle<()>>,
pub client_receiver: Receiver<Client>, pub client_receiver: Receiver<ClientPackage>,
pub info_requester_receiver: Receiver<Sender<ServerInfoPacket>>, pub info_requester_receiver: Receiver<Sender<ServerInfoPacket>>,
stop_sender: Option<oneshot::Sender<()>>, stop_sender: Option<oneshot::Sender<()>>,
} }
@ -31,7 +43,7 @@ impl ConnectionHandler {
let network_clone = Arc::clone(&network); let network_clone = Arc::clone(&network);
let (stop_sender, stop_receiver) = oneshot::channel(); let (stop_sender, stop_receiver) = oneshot::channel();
let (client_sender, client_receiver) = unbounded::<Client>(); let (client_sender, client_receiver) = unbounded::<ClientPackage>();
let (info_requester_sender, info_requester_receiver) = let (info_requester_sender, info_requester_receiver) =
bounded::<Sender<ServerInfoPacket>>(1); bounded::<Sender<ServerInfoPacket>>(1);
@ -55,7 +67,7 @@ impl ConnectionHandler {
async fn work( async fn work(
network: Arc<Network>, network: Arc<Network>,
client_sender: Sender<Client>, client_sender: Sender<ClientPackage>,
info_requester_sender: Sender<Sender<ServerInfoPacket>>, info_requester_sender: Sender<Sender<ServerInfoPacket>>,
stop_receiver: oneshot::Receiver<()>, stop_receiver: oneshot::Receiver<()>,
) { ) {
@ -92,7 +104,7 @@ impl ConnectionHandler {
async fn init_participant( async fn init_participant(
participant: Participant, participant: Participant,
client_sender: Sender<Client>, client_sender: Sender<ClientPackage>,
info_requester_sender: Sender<Sender<ServerInfoPacket>>, info_requester_sender: Sender<Sender<ServerInfoPacket>>,
) -> Result<(), Box<dyn std::error::Error>> { ) -> Result<(), Box<dyn std::error::Error>> {
debug!("New Participant connected to the server"); debug!("New Participant connected to the server");
@ -129,17 +141,20 @@ impl ConnectionHandler {
client_type, client_type,
in_game: None, in_game: None,
participant: Some(participant), participant: Some(participant),
general_stream,
ping_stream,
register_stream,
in_game_stream,
character_screen_stream,
network_error: false,
last_ping: server_data.time, last_ping: server_data.time,
login_msg_sent: false, login_msg_sent: false,
}; };
client_sender.send(client)?; let package = ClientPackage {
client,
general: GeneralStream(general_stream),
ping: PingStream(ping_stream),
register: RegisterStream(register_stream),
character: CharacterScreenStream(character_screen_stream),
in_game: InGameStream(in_game_stream),
};
client_sender.send(package)?;
Ok(()) Ok(())
} }
} }

8
server/src/events/entity_manipulation.rs
View File

@ -1,5 +1,5 @@
use crate::{ use crate::{
client::Client, client::{Client, InGameStream},
comp::{biped_large, quadruped_medium, quadruped_small}, comp::{biped_large, quadruped_medium, quadruped_small},
Server, SpawnPoint, StateExt, Server, SpawnPoint, StateExt,
}; };
@ -42,9 +42,9 @@ pub fn handle_knockback(server: &Server, entity: EcsEntity, impulse: Vec3<f32>)
if let Some(vel) = velocities.get_mut(entity) { if let Some(vel) = velocities.get_mut(entity) {
vel.0 = impulse; vel.0 = impulse;
} }
let mut clients = state.ecs().write_storage::<Client>(); let mut in_game_streams = state.ecs().write_storage::<InGameStream>();
if let Some(client) = clients.get_mut(entity) { if let Some(in_game_stream) = in_game_streams.get_mut(entity) {
client.send_msg(ServerGeneral::Knockback(impulse)); let _ = in_game_stream.0.send(ServerGeneral::Knockback(impulse));
} }
} }

187
server/src/events/group_manip.rs
View File

@ -1,4 +1,7 @@
use crate::{client::Client, Server}; use crate::{
client::{GeneralStream, InGameStream},
Server,
};
use common::{ use common::{
comp::{ comp::{
self, self,
@ -25,20 +28,21 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
match manip { match manip {
GroupManip::Invite(uid) => { GroupManip::Invite(uid) => {
let mut clients = state.ecs().write_storage::<Client>(); let mut general_streams = state.ecs().write_storage::<GeneralStream>();
let invitee = match state.ecs().entity_from_uid(uid.into()) { let invitee =
Some(t) => t, match state.ecs().entity_from_uid(uid.into()) {
None => { Some(t) => t,
// Inform of failure None => {
if let Some(client) = clients.get_mut(entity) { // Inform of failure
client.send_msg( if let Some(general_stream) = general_streams.get_mut(entity) {
ChatType::Meta let _ =
.server_msg("Invite failed, target does not exist.".to_owned()), general_stream.0.send(ChatType::Meta.server_msg(
); "Invite failed, target does not exist.".to_owned(),
} ));
return; }
}, return;
}; },
};
let uids = state.ecs().read_storage::<sync::Uid>(); let uids = state.ecs().read_storage::<sync::Uid>();
@ -62,8 +66,8 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
}); });
if already_in_same_group { if already_in_same_group {
// Inform of failure // Inform of failure
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg(ChatType::Meta.server_msg( let _ = general_stream.0.send(ChatType::Meta.server_msg(
"Invite failed, can't invite someone already in your group".to_owned(), "Invite failed, can't invite someone already in your group".to_owned(),
)); ));
} }
@ -92,8 +96,8 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
>= max_group_size as usize; >= max_group_size as usize;
if group_size_limit_reached { if group_size_limit_reached {
// Inform inviter that they have reached the group size limit // Inform inviter that they have reached the group size limit
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg( let _ = general_stream.0.send(
ChatType::Meta.server_msg( ChatType::Meta.server_msg(
"Invite failed, pending invites plus current group size have reached \ "Invite failed, pending invites plus current group size have reached \
the group size limit" the group size limit"
@ -109,11 +113,13 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
if invites.contains(invitee) { if invites.contains(invitee) {
// Inform inviter that there is already an invite // Inform inviter that there is already an invite
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg( let _ =
ChatType::Meta general_stream
.server_msg("This player already has a pending invite.".to_owned()), .0
); .send(ChatType::Meta.server_msg(
"This player already has a pending invite.".to_owned(),
));
} }
return; return;
} }
@ -150,33 +156,35 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
} }
}; };
let mut in_game_streams = state.ecs().write_storage::<InGameStream>();
// If client comp // If client comp
if let (Some(client), Some(inviter)) = if let (Some(in_game_stream), Some(inviter)) =
(clients.get_mut(invitee), uids.get(entity).copied()) (in_game_streams.get_mut(invitee), uids.get(entity).copied())
{ {
if send_invite() { if send_invite() {
client.send_msg(ServerGeneral::GroupInvite { let _ = in_game_stream.0.send(ServerGeneral::GroupInvite {
inviter, inviter,
timeout: PRESENTED_INVITE_TIMEOUT_DUR, timeout: PRESENTED_INVITE_TIMEOUT_DUR,
}); });
} }
} else if agents.contains(invitee) { } else if agents.contains(invitee) {
send_invite(); send_invite();
} else if let Some(client) = clients.get_mut(entity) { } else if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg( let _ = general_stream.0.send(
ChatType::Meta.server_msg("Can't invite, not a player or npc".to_owned()), ChatType::Meta.server_msg("Can't invite, not a player or npc".to_owned()),
); );
} }
// Notify inviter that the invite is pending // Notify inviter that the invite is pending
if invite_sent { if invite_sent {
if let Some(client) = clients.get_mut(entity) { if let Some(in_game_stream) = in_game_streams.get_mut(entity) {
client.send_msg(ServerGeneral::InvitePending(uid)); let _ = in_game_stream.0.send(ServerGeneral::InvitePending(uid));
} }
} }
}, },
GroupManip::Accept => { GroupManip::Accept => {
let mut clients = state.ecs().write_storage::<Client>(); let mut in_game_streams = state.ecs().write_storage::<InGameStream>();
let uids = state.ecs().read_storage::<sync::Uid>(); let uids = state.ecs().read_storage::<sync::Uid>();
let mut invites = state.ecs().write_storage::<Invite>(); let mut invites = state.ecs().write_storage::<Invite>();
if let Some(inviter) = invites.remove(entity).and_then(|invite| { if let Some(inviter) = invites.remove(entity).and_then(|invite| {
@ -193,13 +201,13 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
Some(inviter) Some(inviter)
}) { }) {
if let (Some(client), Some(target)) = if let (Some(in_game_stream), Some(target)) =
(clients.get_mut(inviter), uids.get(entity).copied()) (in_game_streams.get_mut(inviter), uids.get(entity).copied())
{ {
client.send_msg(ServerGeneral::InviteComplete { let _ = in_game_stream.0.send(ServerGeneral::InviteComplete {
target, target,
answer: InviteAnswer::Accepted, answer: InviteAnswer::Accepted,
}) });
} }
let mut group_manager = state.ecs().write_resource::<GroupManager>(); let mut group_manager = state.ecs().write_resource::<GroupManager>();
group_manager.add_group_member( group_manager.add_group_member(
@ -210,20 +218,20 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
&state.ecs().read_storage(), &state.ecs().read_storage(),
&uids, &uids,
|entity, group_change| { |entity, group_change| {
clients in_game_streams
.get_mut(entity) .get_mut(entity)
.and_then(|c| { .and_then(|s| {
group_change group_change
.try_map(|e| uids.get(e).copied()) .try_map(|e| uids.get(e).copied())
.map(|g| (g, c)) .map(|g| (g, s))
}) })
.map(|(g, c)| c.send_msg(ServerGeneral::GroupUpdate(g))); .map(|(g, s)| s.0.send(ServerGeneral::GroupUpdate(g)));
}, },
); );
} }
}, },
GroupManip::Decline => { GroupManip::Decline => {
let mut clients = state.ecs().write_storage::<Client>(); let mut in_game_streams = state.ecs().write_storage::<InGameStream>();
let uids = state.ecs().read_storage::<sync::Uid>(); let uids = state.ecs().read_storage::<sync::Uid>();
let mut invites = state.ecs().write_storage::<Invite>(); let mut invites = state.ecs().write_storage::<Invite>();
if let Some(inviter) = invites.remove(entity).and_then(|invite| { if let Some(inviter) = invites.remove(entity).and_then(|invite| {
@ -241,18 +249,18 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
Some(inviter) Some(inviter)
}) { }) {
// Inform inviter of rejection // Inform inviter of rejection
if let (Some(client), Some(target)) = if let (Some(in_game_stream), Some(target)) =
(clients.get_mut(inviter), uids.get(entity).copied()) (in_game_streams.get_mut(inviter), uids.get(entity).copied())
{ {
client.send_msg(ServerGeneral::InviteComplete { let _ = in_game_stream.0.send(ServerGeneral::InviteComplete {
target, target,
answer: InviteAnswer::Declined, answer: InviteAnswer::Declined,
}) });
} }
} }
}, },
GroupManip::Leave => { GroupManip::Leave => {
let mut clients = state.ecs().write_storage::<Client>(); let mut in_game_streams = state.ecs().write_storage::<InGameStream>();
let uids = state.ecs().read_storage::<sync::Uid>(); let uids = state.ecs().read_storage::<sync::Uid>();
let mut group_manager = state.ecs().write_resource::<GroupManager>(); let mut group_manager = state.ecs().write_resource::<GroupManager>();
group_manager.leave_group( group_manager.leave_group(
@ -262,19 +270,19 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
&uids, &uids,
&state.ecs().entities(), &state.ecs().entities(),
&mut |entity, group_change| { &mut |entity, group_change| {
clients in_game_streams
.get_mut(entity) .get_mut(entity)
.and_then(|c| { .and_then(|s| {
group_change group_change
.try_map(|e| uids.get(e).copied()) .try_map(|e| uids.get(e).copied())
.map(|g| (g, c)) .map(|g| (g, s))
}) })
.map(|(g, c)| c.send_msg(ServerGeneral::GroupUpdate(g))); .map(|(g, s)| s.0.send(ServerGeneral::GroupUpdate(g)));
}, },
); );
}, },
GroupManip::Kick(uid) => { GroupManip::Kick(uid) => {
let mut clients = state.ecs().write_storage::<Client>(); let mut general_streams = state.ecs().write_storage::<GeneralStream>();
let uids = state.ecs().read_storage::<sync::Uid>(); let uids = state.ecs().read_storage::<sync::Uid>();
let alignments = state.ecs().read_storage::<comp::Alignment>(); let alignments = state.ecs().read_storage::<comp::Alignment>();
@ -282,8 +290,8 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
Some(t) => t, Some(t) => t,
None => { None => {
// Inform of failure // Inform of failure
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg( let _ = general_stream.0.send(
ChatType::Meta ChatType::Meta
.server_msg("Kick failed, target does not exist.".to_owned()), .server_msg("Kick failed, target does not exist.".to_owned()),
); );
@ -295,8 +303,8 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
// Can't kick pet // Can't kick pet
if matches!(alignments.get(target), Some(comp::Alignment::Owned(owner)) if uids.get(target).map_or(true, |u| u != owner)) if matches!(alignments.get(target), Some(comp::Alignment::Owned(owner)) if uids.get(target).map_or(true, |u| u != owner))
{ {
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg( let _ = general_stream.0.send(
ChatType::Meta.server_msg("Kick failed, you can't kick pets.".to_owned()), ChatType::Meta.server_msg("Kick failed, you can't kick pets.".to_owned()),
); );
} }
@ -304,8 +312,8 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
} }
// Can't kick yourself // Can't kick yourself
if uids.get(entity).map_or(false, |u| *u == uid) { if uids.get(entity).map_or(false, |u| *u == uid) {
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg( let _ = general_stream.0.send(
ChatType::Meta ChatType::Meta
.server_msg("Kick failed, you can't kick yourself.".to_owned()), .server_msg("Kick failed, you can't kick yourself.".to_owned()),
); );
@ -315,6 +323,7 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
let mut groups = state.ecs().write_storage::<group::Group>(); let mut groups = state.ecs().write_storage::<group::Group>();
let mut group_manager = state.ecs().write_resource::<GroupManager>(); let mut group_manager = state.ecs().write_resource::<GroupManager>();
let mut in_game_streams = state.ecs().write_storage::<InGameStream>();
// Make sure kicker is the group leader // Make sure kicker is the group leader
match groups match groups
.get(target) .get(target)
@ -329,58 +338,59 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
&uids, &uids,
&state.ecs().entities(), &state.ecs().entities(),
&mut |entity, group_change| { &mut |entity, group_change| {
clients in_game_streams
.get_mut(entity) .get_mut(entity)
.and_then(|c| { .and_then(|s| {
group_change group_change
.try_map(|e| uids.get(e).copied()) .try_map(|e| uids.get(e).copied())
.map(|g| (g, c)) .map(|g| (g, s))
}) })
.map(|(g, c)| c.send_msg(ServerGeneral::GroupUpdate(g))); .map(|(g, s)| s.0.send(ServerGeneral::GroupUpdate(g)));
}, },
); );
// Tell them the have been kicked // Tell them the have been kicked
if let Some(client) = clients.get_mut(target) { if let Some(general_stream) = general_streams.get_mut(target) {
client.send_msg( let _ = general_stream.0.send(
ChatType::Meta ChatType::Meta
.server_msg("You were removed from the group.".to_owned()), .server_msg("You were removed from the group.".to_owned()),
); );
} }
// Tell kicker that they were succesful // Tell kicker that they were succesful
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg(ChatType::Meta.server_msg("Player kicked.".to_owned())); let _ = general_stream
.0
.send(ChatType::Meta.server_msg("Player kicked.".to_owned()));
} }
}, },
Some(_) => { Some(_) => {
// Inform kicker that they are not the leader // Inform kicker that they are not the leader
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg(ChatType::Meta.server_msg( let _ = general_stream.0.send(ChatType::Meta.server_msg(
"Kick failed: You are not the leader of the target's group.".to_owned(), "Kick failed: You are not the leader of the target's group.".to_owned(),
)); ));
} }
}, },
None => { None => {
// Inform kicker that the target is not in a group // Inform kicker that the target is not in a group
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg( let _ =
ChatType::Meta.server_msg( general_stream.0.send(ChatType::Meta.server_msg(
"Kick failed: Your target is not in a group.".to_owned(), "Kick failed: Your target is not in a group.".to_owned(),
), ));
);
} }
}, },
} }
}, },
GroupManip::AssignLeader(uid) => { GroupManip::AssignLeader(uid) => {
let mut clients = state.ecs().write_storage::<Client>(); let mut general_streams = state.ecs().write_storage::<GeneralStream>();
let uids = state.ecs().read_storage::<sync::Uid>(); let uids = state.ecs().read_storage::<sync::Uid>();
let target = match state.ecs().entity_from_uid(uid.into()) { let target = match state.ecs().entity_from_uid(uid.into()) {
Some(t) => t, Some(t) => t,
None => { None => {
// Inform of failure // Inform of failure
if let Some(client) = clients.get_mut(entity) { if let Some(general_stream) = general_streams.get_mut(entity) {
client.send_msg(ChatType::Meta.server_msg( let _ = general_stream.0.send(ChatType::Meta.server_msg(
"Leadership transfer failed, target does not exist".to_owned(), "Leadership transfer failed, target does not exist".to_owned(),
)); ));
} }
@ -389,6 +399,7 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
}; };
let groups = state.ecs().read_storage::<group::Group>(); let groups = state.ecs().read_storage::<group::Group>();
let mut group_manager = state.ecs().write_resource::<GroupManager>(); let mut group_manager = state.ecs().write_resource::<GroupManager>();
let mut in_game_streams = state.ecs().write_storage::<InGameStream>();
// Make sure assigner is the group leader // Make sure assigner is the group leader
match groups match groups
.get(target) .get(target)
@ -403,25 +414,25 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
&state.ecs().read_storage(), &state.ecs().read_storage(),
&uids, &uids,
|entity, group_change| { |entity, group_change| {
clients in_game_streams
.get_mut(entity) .get_mut(entity)
.and_then(|c| { .and_then(|s| {
group_change group_change
.try_map(|e| uids.get(e).copied()) .try_map(|e| uids.get(e).copied())
.map(|g| (g, c)) .map(|g| (g, s))
}) })
.map(|(g, c)| c.send_msg(ServerGeneral::GroupUpdate(g))); .map(|(g, s)| s.0.send(ServerGeneral::GroupUpdate(g)));
}, },
); );
// Tell them they are the leader // Tell them they are the leader
if let Some(client) = clients.get_mut(target) { if let Some(general_stream) = general_streams.get_mut(target) {
client.send_msg( let _ = general_stream.0.send(
ChatType::Meta.server_msg("You are the group leader now.".to_owned()), ChatType::Meta.server_msg("You are the group leader now.".to_owned()),
); );
} }
// Tell the old leader that the transfer was succesful // Tell the old leader that the transfer was succesful
if let Some(client) = clients.get_mut(target) { if let Some(general_stream) = general_streams.get_mut(target) {
client.send_msg( let _ = general_stream.0.send(
ChatType::Meta ChatType::Meta
.server_msg("You are no longer the group leader.".to_owned()), .server_msg("You are no longer the group leader.".to_owned()),
); );
@ -429,8 +440,9 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
}, },
Some(_) => { Some(_) => {
// Inform transferer that they are not the leader // Inform transferer that they are not the leader
if let Some(client) = clients.get_mut(entity) { let mut general_streams = state.ecs().write_storage::<GeneralStream>();
client.send_msg( if let Some(general_stream) = general_streams.get_mut(entity) {
let _ = general_stream.0.send(
ChatType::Meta.server_msg( ChatType::Meta.server_msg(
"Transfer failed: You are not the leader of the target's group." "Transfer failed: You are not the leader of the target's group."
.to_owned(), .to_owned(),
@ -440,8 +452,9 @@ pub fn handle_group(server: &mut Server, entity: specs::Entity, manip: GroupMani
}, },
None => { None => {
// Inform transferer that the target is not in a group // Inform transferer that the target is not in a group
if let Some(client) = clients.get_mut(entity) { let mut general_streams = state.ecs().write_storage::<GeneralStream>();
client.send_msg(ChatType::Meta.server_msg( if let Some(general_stream) = general_streams.get_mut(entity) {
let _ = general_stream.0.send(ChatType::Meta.server_msg(
"Transfer failed: Your target is not in a group.".to_owned(), "Transfer failed: Your target is not in a group.".to_owned(),
)); ));
} }

214
server/src/events/interaction.rs
View File

@ -1,5 +1,8 @@
use crate::{ use crate::{
client::{Client, RegionSubscription}, client::{
CharacterScreenStream, Client, GeneralStream, InGameStream, PingStream, RegionSubscription,
RegisterStream,
},
Server, Server,
}; };
use common::{ use common::{
@ -120,80 +123,153 @@ pub fn handle_possess(server: &Server, possessor_uid: Uid, possesse_uid: Uid) {
// You can't possess other players // You can't possess other players
let mut clients = ecs.write_storage::<Client>(); let mut clients = ecs.write_storage::<Client>();
let mut general_streams = ecs.write_storage::<GeneralStream>();
let mut ping_streams = ecs.write_storage::<PingStream>();
let mut register_streams = ecs.write_storage::<RegisterStream>();
let mut character_screen_streams = ecs.write_storage::<CharacterScreenStream>();
let mut in_game_streams = ecs.write_storage::<InGameStream>();
if clients.get_mut(possesse).is_none() { if clients.get_mut(possesse).is_none() {
if let Some(mut client) = clients.remove(possessor) { let client = match clients.remove(possessor) {
client.send_msg(ServerGeneral::SetPlayerEntity(possesse_uid)); Some(c) => c,
clients None => return,
.insert(possesse, client) };
.err() let mut general_stream = match general_streams.remove(possessor) {
.map(|e| error!(?e, "Error inserting client component during possession")); Some(c) => c,
// Put possess item into loadout None => return,
let mut loadouts = ecs.write_storage::<comp::Loadout>(); };
let loadout = loadouts let ping_stream = match ping_streams.remove(possessor) {
.entry(possesse) Some(c) => c,
.expect("Could not read loadouts component while possessing") None => return,
.or_insert(comp::Loadout::default()); };
let register_stream = match register_streams.remove(possessor) {
Some(c) => c,
None => return,
};
let character_screen_stream = match character_screen_streams.remove(possessor) {
Some(c) => c,
None => return,
};
let in_game_stream = match in_game_streams.remove(possessor) {
Some(c) => c,
None => return,
};
let _ = general_stream
.0
.send(ServerGeneral::SetPlayerEntity(possesse_uid));
clients
.insert(possesse, client)
.err()
.map(|e| error!(?e, "Error inserting client component during possession"));
general_streams
.insert(possesse, general_stream)
.err()
.map(|e| {
error!(
?e,
"Error inserting general_streams component during possession"
)
});
ping_streams.insert(possesse, ping_stream).err().map(|e| {
error!(
?e,
"Error inserting ping_streams component during possession"
)
});
register_streams
.insert(possesse, register_stream)
.err()
.map(|e| {
error!(
?e,
"Error inserting register_streams component during possession"
)
});
character_screen_streams
.insert(possesse, character_screen_stream)
.err()
.map(|e| {
error!(
?e,
"Error inserting character_screen_streams component during possession"
)
});
in_game_streams
.insert(possesse, in_game_stream)
.err()
.map(|e| {
error!(
?e,
"Error inserting in_game_streams component during possession"
)
});
// Put possess item into loadout
let mut loadouts = ecs.write_storage::<comp::Loadout>();
let loadout = loadouts
.entry(possesse)
.expect("Could not read loadouts component while possessing")
.or_insert(comp::Loadout::default());
let item = comp::Item::new_from_asset_expect("common.items.debug.possess"); let item = comp::Item::new_from_asset_expect("common.items.debug.possess");
if let item::ItemKind::Tool(tool) = item.kind() { if let item::ItemKind::Tool(tool) = item.kind() {
let mut abilities = tool.get_abilities(); let mut abilities = tool.get_abilities();
let mut ability_drain = abilities.drain(..); let mut ability_drain = abilities.drain(..);
let debug_item = comp::ItemConfig { let debug_item = comp::ItemConfig {
item, item,
ability1: ability_drain.next(), ability1: ability_drain.next(),
ability2: ability_drain.next(), ability2: ability_drain.next(),
ability3: ability_drain.next(), ability3: ability_drain.next(),
block_ability: None, block_ability: None,
dodge_ability: None, dodge_ability: None,
}; };
std::mem::swap(&mut loadout.active_item, &mut loadout.second_item); std::mem::swap(&mut loadout.active_item, &mut loadout.second_item);
loadout.active_item = Some(debug_item); loadout.active_item = Some(debug_item);
} }
// Move player component // Move player component
{ {
let mut players = ecs.write_storage::<comp::Player>(); let mut players = ecs.write_storage::<comp::Player>();
if let Some(player) = players.remove(possessor) { if let Some(player) = players.remove(possessor) {
players.insert(possesse, player).err().map(|e| { players
error!(?e, "Error inserting player component during possession") .insert(possesse, player)
}); .err()
} .map(|e| error!(?e, "Error inserting player component during possession"));
} }
// Transfer region subscription }
{ // Transfer region subscription
let mut subscriptions = ecs.write_storage::<RegionSubscription>(); {
if let Some(s) = subscriptions.remove(possessor) { let mut subscriptions = ecs.write_storage::<RegionSubscription>();
subscriptions.insert(possesse, s).err().map(|e| { if let Some(s) = subscriptions.remove(possessor) {
error!( subscriptions.insert(possesse, s).err().map(|e| {
?e, error!(
"Error inserting subscription component during possession" ?e,
) "Error inserting subscription component during possession"
}); )
} });
} }
// Remove will of the entity }
ecs.write_storage::<comp::Agent>().remove(possesse); // Remove will of the entity
// Reset controller of former shell ecs.write_storage::<comp::Agent>().remove(possesse);
ecs.write_storage::<comp::Controller>() // Reset controller of former shell
.get_mut(possessor) ecs.write_storage::<comp::Controller>()
.map(|c| c.reset()); .get_mut(possessor)
// Transfer admin powers .map(|c| c.reset());
{ // Transfer admin powers
let mut admins = ecs.write_storage::<comp::Admin>(); {
if let Some(admin) = admins.remove(possessor) { let mut admins = ecs.write_storage::<comp::Admin>();
admins.insert(possesse, admin).err().map(|e| { if let Some(admin) = admins.remove(possessor) {
error!(?e, "Error inserting admin component during possession") admins
}); .insert(possesse, admin)
} .err()
.map(|e| error!(?e, "Error inserting admin component during possession"));
} }
// Transfer waypoint }
{ // Transfer waypoint
let mut waypoints = ecs.write_storage::<comp::Waypoint>(); {
if let Some(waypoint) = waypoints.remove(possessor) { let mut waypoints = ecs.write_storage::<comp::Waypoint>();
waypoints.insert(possesse, waypoint).err().map(|e| { if let Some(waypoint) = waypoints.remove(possessor) {
error!(?e, "Error inserting waypoint component during possession",) waypoints.insert(possesse, waypoint).err().map(|e| {
}); error!(?e, "Error inserting waypoint component during possession",)
} });
} }
} }
} }

15
server/src/events/inventory_manip.rs
View File

@ -1,4 +1,4 @@
use crate::{client::Client, Server, StateExt}; use crate::{client::InGameStream, Server, StateExt};
use common::{ use common::{
comp::{ comp::{
self, item, self, item,
@ -279,7 +279,8 @@ pub fn handle_inventory(server: &mut Server, entity: EcsEntity, manip: comp::Inv
.insert(tameable_entity, comp::Alignment::Owned(uid)); .insert(tameable_entity, comp::Alignment::Owned(uid));
// Add to group system // Add to group system
let mut clients = state.ecs().write_storage::<Client>(); let mut in_game_streams =
state.ecs().write_storage::<InGameStream>();
let uids = state.ecs().read_storage::<Uid>(); let uids = state.ecs().read_storage::<Uid>();
let mut group_manager = state let mut group_manager = state
.ecs() .ecs()
@ -293,15 +294,15 @@ pub fn handle_inventory(server: &mut Server, entity: EcsEntity, manip: comp::Inv
&state.ecs().read_storage(), &state.ecs().read_storage(),
&uids, &uids,
&mut |entity, group_change| { &mut |entity, group_change| {
clients in_game_streams
.get_mut(entity) .get_mut(entity)
.and_then(|c| { .and_then(|s| {
group_change group_change
.try_map(|e| uids.get(e).copied()) .try_map(|e| uids.get(e).copied())
.map(|g| (g, c)) .map(|g| (g, s))
}) })
.map(|(g, c)| { .map(|(g, s)| {
c.send_msg(ServerGeneral::GroupUpdate(g)) s.0.send(ServerGeneral::GroupUpdate(g))
}); });
}, },
); );

53
server/src/events/player.rs
View File

@ -1,6 +1,12 @@
use super::Event; use super::Event;
use crate::{ use crate::{
client::Client, login_provider::LoginProvider, persistence, state_ext::StateExt, Server, client::{
CharacterScreenStream, Client, GeneralStream, InGameStream, PingStream, RegisterStream,
},
login_provider::LoginProvider,
persistence,
state_ext::StateExt,
Server,
}; };
use common::{ use common::{
comp, comp,
@ -17,26 +23,61 @@ pub fn handle_exit_ingame(server: &mut Server, entity: EcsEntity) {
span!(_guard, "handle_exit_ingame"); span!(_guard, "handle_exit_ingame");
let state = server.state_mut(); let state = server.state_mut();
// Create new entity with just `Client`, `Uid`, and `Player` components // Create new entity with just `Client`, `Uid`, `Player`, and `...Stream`
// Easier than checking and removing all other known components // components Easier than checking and removing all other known components
// Note: If other `ServerEvent`s are referring to this entity they will be // Note: If other `ServerEvent`s are referring to this entity they will be
// disrupted // disrupted
let maybe_client = state.ecs().write_storage::<Client>().remove(entity); let maybe_client = state.ecs().write_storage::<Client>().remove(entity);
let maybe_uid = state.read_component_copied::<Uid>(entity); let maybe_uid = state.read_component_copied::<Uid>(entity);
let maybe_player = state.ecs().write_storage::<comp::Player>().remove(entity); let maybe_player = state.ecs().write_storage::<comp::Player>().remove(entity);
let maybe_admin = state.ecs().write_storage::<comp::Admin>().remove(entity); let maybe_admin = state.ecs().write_storage::<comp::Admin>().remove(entity);
let maybe_general_stream = state.ecs().write_storage::<GeneralStream>().remove(entity);
let maybe_ping_stream = state.ecs().write_storage::<PingStream>().remove(entity);
let maybe_register_stream = state.ecs().write_storage::<RegisterStream>().remove(entity);
let maybe_character_screen_stream = state
.ecs()
.write_storage::<CharacterScreenStream>()
.remove(entity);
let maybe_in_game_stream = state.ecs().write_storage::<InGameStream>().remove(entity);
let maybe_group = state let maybe_group = state
.ecs() .ecs()
.write_storage::<group::Group>() .write_storage::<group::Group>()
.get(entity) .get(entity)
.cloned(); .cloned();
if let (Some(mut client), Some(uid), Some(player)) = (maybe_client, maybe_uid, maybe_player) { if let (
Some(mut client),
Some(uid),
Some(player),
Some(mut general_stream),
Some(ping_stream),
Some(register_stream),
Some(character_screen_stream),
Some(in_game_stream),
) = (
maybe_client,
maybe_uid,
maybe_player,
maybe_general_stream,
maybe_ping_stream,
maybe_register_stream,
maybe_character_screen_stream,
maybe_in_game_stream,
) {
// Tell client its request was successful // Tell client its request was successful
client.in_game = None; client.in_game = None;
client.send_msg(ServerGeneral::ExitInGameSuccess); let _ = general_stream.0.send(ServerGeneral::ExitInGameSuccess);
let entity_builder = state.ecs_mut().create_entity().with(client).with(player); let entity_builder = state
.ecs_mut()
.create_entity()
.with(client)
.with(player)
.with(general_stream)
.with(ping_stream)
.with(register_stream)
.with(character_screen_stream)
.with(in_game_stream);
// Preserve group component if present // Preserve group component if present
let entity_builder = match maybe_group { let entity_builder = match maybe_group {

93
server/src/lib.rs
View File

@ -34,7 +34,10 @@ pub use crate::{
use crate::{ use crate::{
alias_validator::AliasValidator, alias_validator::AliasValidator,
chunk_generator::ChunkGenerator, chunk_generator::ChunkGenerator,
client::{Client, RegionSubscription}, client::{
CharacterScreenStream, Client, GeneralStream, InGameStream, PingStream, RegionSubscription,
RegisterStream,
},
cmd::ChatCommandExt, cmd::ChatCommandExt,
connection_handler::ConnectionHandler, connection_handler::ConnectionHandler,
data_dir::DataDir, data_dir::DataDir,
@ -180,6 +183,11 @@ impl Server {
// Server-only components // Server-only components
state.ecs_mut().register::<RegionSubscription>(); state.ecs_mut().register::<RegionSubscription>();
state.ecs_mut().register::<Client>(); state.ecs_mut().register::<Client>();
state.ecs_mut().register::<GeneralStream>();
state.ecs_mut().register::<PingStream>();
state.ecs_mut().register::<RegisterStream>();
state.ecs_mut().register::<CharacterScreenStream>();
state.ecs_mut().register::<InGameStream>();
//Alias validator //Alias validator
let banned_words_paths = &settings.banned_words_files; let banned_words_paths = &settings.banned_words_files;
@ -804,8 +812,8 @@ impl Server {
}); });
} }
while let Ok(data) = self.connection_handler.client_receiver.try_recv() { while let Ok(mut package) = self.connection_handler.client_receiver.try_recv() {
let mut client = data; let client = package.client;
if self.settings().max_players if self.settings().max_players
<= self.state.ecs().read_storage::<Client>().join().count() <= self.state.ecs().read_storage::<Client>().join().count()
@ -814,7 +822,7 @@ impl Server {
?client.participant, ?client.participant,
"to many players, wont allow participant to connect" "to many players, wont allow participant to connect"
); );
client.register_stream.send(ServerInit::TooManyPlayers)?; package.register.0.send(ServerInit::TooManyPlayers)?;
continue; continue;
} }
@ -823,6 +831,11 @@ impl Server {
.ecs_mut() .ecs_mut()
.create_entity_synced() .create_entity_synced()
.with(client) .with(client)
.with(package.general)
.with(package.ping)
.with(package.register)
.with(package.character)
.with(package.in_game)
.build(); .build();
self.state self.state
.ecs() .ecs()
@ -834,10 +847,10 @@ impl Server {
debug!("Starting initial sync with client."); debug!("Starting initial sync with client.");
self.state self.state
.ecs() .ecs()
.write_storage::<Client>() .write_storage::<RegisterStream>()
.get_mut(entity) .get_mut(entity)
.unwrap() .unwrap()
.register_stream .0
.send(ServerInit::GameSync { .send(ServerInit::GameSync {
// Send client their entity // Send client their entity
entity_package: TrackedComps::fetch(&self.state.ecs()) entity_package: TrackedComps::fetch(&self.state.ecs())
@ -859,8 +872,72 @@ impl Server {
where where
S: Into<ServerMsg>, S: Into<ServerMsg>,
{ {
if let Some(client) = self.state.ecs().write_storage::<Client>().get_mut(entity) { const ERR: &str =
client.send_msg(msg.into()) "Don't do that. Sending these messages is only done ONCE at connect and not by this fn";
match msg.into() {
ServerMsg::Info(_) => panic!(ERR),
ServerMsg::Init(_) => panic!(ERR),
ServerMsg::RegisterAnswer(msg) => {
self.state
.ecs()
.write_storage::<RegisterStream>()
.get_mut(entity)
.map(|s| s.0.send(msg));
},
ServerMsg::General(msg) => {
match &msg {
//Character Screen related
ServerGeneral::CharacterDataLoadError(_)
| ServerGeneral::CharacterListUpdate(_)
| ServerGeneral::CharacterActionError(_)
| ServerGeneral::CharacterSuccess => self
.state
.ecs()
.write_storage::<CharacterScreenStream>()
.get_mut(entity)
.map(|s| s.0.send(msg)),
//Ingame related
ServerGeneral::GroupUpdate(_)
| ServerGeneral::GroupInvite { .. }
| ServerGeneral::InvitePending(_)
| ServerGeneral::InviteComplete { .. }
| ServerGeneral::ExitInGameSuccess
| ServerGeneral::InventoryUpdate(_, _)
| ServerGeneral::TerrainChunkUpdate { .. }
| ServerGeneral::TerrainBlockUpdates(_)
| ServerGeneral::SetViewDistance(_)
| ServerGeneral::Outcomes(_)
| ServerGeneral::Knockback(_) => self
.state
.ecs()
.write_storage::<InGameStream>()
.get_mut(entity)
.map(|s| s.0.send(msg)),
// Always possible
ServerGeneral::PlayerListUpdate(_)
| ServerGeneral::ChatMsg(_)
| ServerGeneral::SetPlayerEntity(_)
| ServerGeneral::TimeOfDay(_)
| ServerGeneral::EntitySync(_)
| ServerGeneral::CompSync(_)
| ServerGeneral::CreateEntity(_)
| ServerGeneral::DeleteEntity(_)
| ServerGeneral::Disconnect(_)
| ServerGeneral::Notification(_) => self
.state
.ecs()
.write_storage::<GeneralStream>()
.get_mut(entity)
.map(|s| s.0.send(msg)),
};
},
ServerMsg::Ping(msg) => {
self.state
.ecs()
.write_storage::<PingStream>()
.get_mut(entity)
.map(|s| s.0.send(msg));
},
} }
} }

97
server/src/state_ext.rs
View File

@ -1,11 +1,14 @@
use crate::{ use crate::{
client::Client, persistence::PersistedComponents, sys::sentinel::DeletedEntities, SpawnPoint, client::{CharacterScreenStream, Client, GeneralStream},
persistence::PersistedComponents,
sys::sentinel::DeletedEntities,
SpawnPoint,
}; };
use common::{ use common::{
character::CharacterId, character::CharacterId,
comp, comp,
effect::Effect, effect::Effect,
msg::{CharacterInfo, ClientInGame, PlayerListUpdate, ServerGeneral, ServerMsg}, msg::{CharacterInfo, ClientInGame, PlayerListUpdate, ServerGeneral},
state::State, state::State,
sync::{Uid, UidAllocator, WorldSyncExt}, sync::{Uid, UidAllocator, WorldSyncExt},
util::Dir, util::Dir,
@ -222,8 +225,16 @@ impl StateExt for State {
// Tell the client its request was successful. // Tell the client its request was successful.
if let Some(client) = self.ecs().write_storage::<Client>().get_mut(entity) { if let Some(client) = self.ecs().write_storage::<Client>().get_mut(entity) {
client.in_game = Some(ClientInGame::Character); if let Some(character_screen_stream) = self
client.send_msg(ServerGeneral::CharacterSuccess) .ecs()
.write_storage::<CharacterScreenStream>()
.get_mut(entity)
{
client.in_game = Some(ClientInGame::Character);
let _ = character_screen_stream
.0
.send(ServerGeneral::CharacterSuccess);
}
} }
} }
@ -293,14 +304,16 @@ impl StateExt for State {
} }
}, },
comp::ChatType::Tell(u, t) => { comp::ChatType::Tell(u, t) => {
for (client, uid) in ( for (general_stream, uid) in (
&mut ecs.write_storage::<Client>(), &mut ecs.write_storage::<GeneralStream>(),
&ecs.read_storage::<Uid>(), &ecs.read_storage::<Uid>(),
) )
.join() .join()
{ {
if uid == u || uid == t { if uid == u || uid == t {
client.send_msg(ServerGeneral::ChatMsg(resolved_msg.clone())); let _ = general_stream
.0
.send(ServerGeneral::ChatMsg(resolved_msg.clone()));
} }
} }
}, },
@ -310,9 +323,13 @@ impl StateExt for State {
(*ecs.read_resource::<UidAllocator>()).retrieve_entity_internal(uid.0); (*ecs.read_resource::<UidAllocator>()).retrieve_entity_internal(uid.0);
let positions = ecs.read_storage::<comp::Pos>(); let positions = ecs.read_storage::<comp::Pos>();
if let Some(speaker_pos) = entity_opt.and_then(|e| positions.get(e)) { if let Some(speaker_pos) = entity_opt.and_then(|e| positions.get(e)) {
for (client, pos) in (&mut ecs.write_storage::<Client>(), &positions).join() { for (general_stream, pos) in
(&mut ecs.write_storage::<GeneralStream>(), &positions).join()
{
if is_within(comp::ChatMsg::SAY_DISTANCE, pos, speaker_pos) { if is_within(comp::ChatMsg::SAY_DISTANCE, pos, speaker_pos) {
client.send_msg(ServerGeneral::ChatMsg(resolved_msg.clone())); let _ = general_stream
.0
.send(ServerGeneral::ChatMsg(resolved_msg.clone()));
} }
} }
} }
@ -322,9 +339,13 @@ impl StateExt for State {
(*ecs.read_resource::<UidAllocator>()).retrieve_entity_internal(uid.0); (*ecs.read_resource::<UidAllocator>()).retrieve_entity_internal(uid.0);
let positions = ecs.read_storage::<comp::Pos>(); let positions = ecs.read_storage::<comp::Pos>();
if let Some(speaker_pos) = entity_opt.and_then(|e| positions.get(e)) { if let Some(speaker_pos) = entity_opt.and_then(|e| positions.get(e)) {
for (client, pos) in (&mut ecs.write_storage::<Client>(), &positions).join() { for (general_stream, pos) in
(&mut ecs.write_storage::<GeneralStream>(), &positions).join()
{
if is_within(comp::ChatMsg::REGION_DISTANCE, pos, speaker_pos) { if is_within(comp::ChatMsg::REGION_DISTANCE, pos, speaker_pos) {
client.send_msg(ServerGeneral::ChatMsg(resolved_msg.clone())); let _ = general_stream
.0
.send(ServerGeneral::ChatMsg(resolved_msg.clone()));
} }
} }
} }
@ -334,35 +355,43 @@ impl StateExt for State {
(*ecs.read_resource::<UidAllocator>()).retrieve_entity_internal(uid.0); (*ecs.read_resource::<UidAllocator>()).retrieve_entity_internal(uid.0);
let positions = ecs.read_storage::<comp::Pos>(); let positions = ecs.read_storage::<comp::Pos>();
if let Some(speaker_pos) = entity_opt.and_then(|e| positions.get(e)) { if let Some(speaker_pos) = entity_opt.and_then(|e| positions.get(e)) {
for (client, pos) in (&mut ecs.write_storage::<Client>(), &positions).join() { for (general_stream, pos) in
(&mut ecs.write_storage::<GeneralStream>(), &positions).join()
{
if is_within(comp::ChatMsg::NPC_DISTANCE, pos, speaker_pos) { if is_within(comp::ChatMsg::NPC_DISTANCE, pos, speaker_pos) {
client.send_msg(ServerGeneral::ChatMsg(resolved_msg.clone())); let _ = general_stream
.0
.send(ServerGeneral::ChatMsg(resolved_msg.clone()));
} }
} }
} }
}, },
comp::ChatType::FactionMeta(s) | comp::ChatType::Faction(_, s) => { comp::ChatType::FactionMeta(s) | comp::ChatType::Faction(_, s) => {
for (client, faction) in ( for (general_stream, faction) in (
&mut ecs.write_storage::<Client>(), &mut ecs.write_storage::<GeneralStream>(),
&ecs.read_storage::<comp::Faction>(), &ecs.read_storage::<comp::Faction>(),
) )
.join() .join()
{ {
if s == &faction.0 { if s == &faction.0 {
client.send_msg(ServerGeneral::ChatMsg(resolved_msg.clone())); let _ = general_stream
.0
.send(ServerGeneral::ChatMsg(resolved_msg.clone()));
} }
} }
}, },
comp::ChatType::GroupMeta(g) | comp::ChatType::Group(_, g) => { comp::ChatType::GroupMeta(g) | comp::ChatType::Group(_, g) => {
for (client, group) in ( for (general_stream, group) in (
&mut ecs.write_storage::<Client>(), &mut ecs.write_storage::<GeneralStream>(),
&ecs.read_storage::<comp::Group>(), &ecs.read_storage::<comp::Group>(),
) )
.join() .join()
{ {
if g == group { if g == group {
client.send_msg(ServerGeneral::ChatMsg(resolved_msg.clone())); let _ = general_stream
.0
.send(ServerGeneral::ChatMsg(resolved_msg.clone()));
} }
} }
}, },
@ -371,23 +400,27 @@ impl StateExt for State {
/// Sends the message to all connected clients /// Sends the message to all connected clients
fn notify_registered_clients(&self, msg: ServerGeneral) { fn notify_registered_clients(&self, msg: ServerGeneral) {
let msg: ServerMsg = msg.into(); for (general_stream, _) in (
for client in (&mut self.ecs().write_storage::<Client>()) &mut self.ecs().write_storage::<GeneralStream>(),
&self.ecs().read_storage::<Client>(),
)
.join() .join()
.filter(|c| c.registered) .filter(|(_, c)| c.registered)
{ {
client.send_msg(msg.clone()); let _ = general_stream.0.send(msg.clone());
} }
} }
/// Sends the message to all clients playing in game /// Sends the message to all clients playing in game
fn notify_in_game_clients(&self, msg: ServerGeneral) { fn notify_in_game_clients(&self, msg: ServerGeneral) {
let msg: ServerMsg = msg.into(); for (general_stream, _) in (
for client in (&mut self.ecs().write_storage::<Client>()) &mut self.ecs().write_storage::<GeneralStream>(),
&self.ecs().read_storage::<Client>(),
)
.join() .join()
.filter(|c| c.in_game.is_some()) .filter(|(_, c)| c.in_game.is_some())
{ {
client.send_msg(msg.clone()); let _ = general_stream.0.send(msg.clone());
} }
} }
@ -397,7 +430,7 @@ impl StateExt for State {
) -> Result<(), specs::error::WrongGeneration> { ) -> Result<(), specs::error::WrongGeneration> {
// Remove entity from a group if they are in one // Remove entity from a group if they are in one
{ {
let mut clients = self.ecs().write_storage::<Client>(); let mut general_streams = self.ecs().write_storage::<GeneralStream>();
let uids = self.ecs().read_storage::<Uid>(); let uids = self.ecs().read_storage::<Uid>();
let mut group_manager = self.ecs().write_resource::<comp::group::GroupManager>(); let mut group_manager = self.ecs().write_resource::<comp::group::GroupManager>();
group_manager.entity_deleted( group_manager.entity_deleted(
@ -407,14 +440,14 @@ impl StateExt for State {
&uids, &uids,
&self.ecs().entities(), &self.ecs().entities(),
&mut |entity, group_change| { &mut |entity, group_change| {
clients general_streams
.get_mut(entity) .get_mut(entity)
.and_then(|c| { .and_then(|s| {
group_change group_change
.try_map(|e| uids.get(e).copied()) .try_map(|e| uids.get(e).copied())
.map(|g| (g, c)) .map(|g| (g, s))
}) })
.map(|(g, c)| c.send_msg(ServerGeneral::GroupUpdate(g))); .map(|(g, s)| s.0.send(ServerGeneral::GroupUpdate(g)));
}, },
); );
} }

113
server/src/sys/entity_sync.rs
View File

@ -3,7 +3,7 @@ use super::{
SysTimer, SysTimer,
}; };
use crate::{ use crate::{
client::{Client, RegionSubscription}, client::{Client, GeneralStream, InGameStream, RegionSubscription},
Tick, Tick,
}; };
use common::{ use common::{
@ -43,6 +43,8 @@ impl<'a> System<'a> for Sys {
WriteStorage<'a, Last<Vel>>, WriteStorage<'a, Last<Vel>>,
WriteStorage<'a, Last<Ori>>, WriteStorage<'a, Last<Ori>>,
WriteStorage<'a, Client>, WriteStorage<'a, Client>,
WriteStorage<'a, InGameStream>,
WriteStorage<'a, GeneralStream>,
WriteStorage<'a, ForceUpdate>, WriteStorage<'a, ForceUpdate>,
WriteStorage<'a, InventoryUpdate>, WriteStorage<'a, InventoryUpdate>,
Write<'a, DeletedEntities>, Write<'a, DeletedEntities>,
@ -70,6 +72,8 @@ impl<'a> System<'a> for Sys {
mut last_vel, mut last_vel,
mut last_ori, mut last_ori,
mut clients, mut clients,
mut in_game_streams,
mut general_streams,
mut force_updates, mut force_updates,
mut inventory_updates, mut inventory_updates,
mut deleted_entities, mut deleted_entities,
@ -104,15 +108,31 @@ impl<'a> System<'a> for Sys {
for (key, region) in region_map.iter() { for (key, region) in region_map.iter() {
// Assemble subscriber list for this region by iterating through clients and // Assemble subscriber list for this region by iterating through clients and
// checking if they are subscribed to this region // checking if they are subscribed to this region
let mut subscribers = (&mut clients, &entities, &subscriptions, &positions) let mut subscribers = (
&mut clients,
&entities,
&subscriptions,
&positions,
&mut in_game_streams,
&mut general_streams,
)
.join() .join()
.filter_map(|(client, entity, subscription, pos)| { .filter_map(
if client.in_game.is_some() && subscription.regions.contains(&key) { |(client, entity, subscription, pos, in_game_stream, general_stream)| {
Some((client, &subscription.regions, entity, *pos)) if client.in_game.is_some() && subscription.regions.contains(&key) {
} else { Some((
None client,
} &subscription.regions,
}) entity,
*pos,
in_game_stream,
general_stream,
))
} else {
None
}
},
)
.collect::<Vec<_>>(); .collect::<Vec<_>>();
for event in region.events() { for event in region.events() {
@ -135,7 +155,9 @@ impl<'a> System<'a> for Sys {
vel.copied(), vel.copied(),
ori.copied(), ori.copied(),
)); ));
for (client, regions, client_entity, _) in &mut subscribers { for (_, regions, client_entity, _, _, general_stream) in
&mut subscribers
{
if maybe_key if maybe_key
.as_ref() .as_ref()
.map(|key| !regions.contains(key)) .map(|key| !regions.contains(key))
@ -143,7 +165,7 @@ impl<'a> System<'a> for Sys {
// Client doesn't need to know about itself // Client doesn't need to know about itself
&& *client_entity != entity && *client_entity != entity
{ {
client.send_msg(create_msg.clone()); let _ = general_stream.0.send(create_msg.clone());
} }
} }
} }
@ -151,13 +173,13 @@ impl<'a> System<'a> for Sys {
RegionEvent::Left(id, maybe_key) => { RegionEvent::Left(id, maybe_key) => {
// Lookup UID for entity // Lookup UID for entity
if let Some(&uid) = uids.get(entities.entity(*id)) { if let Some(&uid) = uids.get(entities.entity(*id)) {
for (client, regions, _, _) in &mut subscribers { for (_, regions, _, _, _, general_stream) in &mut subscribers {
if maybe_key if maybe_key
.as_ref() .as_ref()
.map(|key| !regions.contains(key)) .map(|key| !regions.contains(key))
.unwrap_or(true) .unwrap_or(true)
{ {
client.send_msg(ServerGeneral::DeleteEntity(uid)); let _ = general_stream.0.send(ServerGeneral::DeleteEntity(uid));
} }
} }
} }
@ -176,17 +198,19 @@ impl<'a> System<'a> for Sys {
); );
let entity_sync_msg = ServerGeneral::EntitySync(entity_sync_package); let entity_sync_msg = ServerGeneral::EntitySync(entity_sync_package);
let comp_sync_msg = ServerGeneral::CompSync(comp_sync_package); let comp_sync_msg = ServerGeneral::CompSync(comp_sync_package);
subscribers.iter_mut().for_each(move |(client, _, _, _)| { subscribers
client.send_msg(entity_sync_msg.clone()); .iter_mut()
client.send_msg(comp_sync_msg.clone()); .for_each(move |(_, _, _, _, _, general_stream)| {
}); let _ = general_stream.0.send(entity_sync_msg.clone());
let _ = general_stream.0.send(comp_sync_msg.clone());
});
let mut send_msg = |msg: ServerGeneral, let mut send_general = |msg: ServerGeneral,
entity: EcsEntity, entity: EcsEntity,
pos: Pos, pos: Pos,
force_update: Option<&ForceUpdate>, force_update: Option<&ForceUpdate>,
throttle: bool| { throttle: bool| {
for (client, _, client_entity, client_pos) in &mut subscribers { for (_, _, client_entity, client_pos, _, general_stream) in &mut subscribers {
if if client_entity == &entity { if if client_entity == &entity {
// Don't send client physics updates about itself unless force update is set // Don't send client physics updates about itself unless force update is set
force_update.is_some() force_update.is_some()
@ -212,7 +236,7 @@ impl<'a> System<'a> for Sys {
true // Closer than 100 blocks true // Closer than 100 blocks
} }
} { } {
client.send_msg(msg.clone()); let _ = general_stream.0.send(msg.clone());
} }
} }
}; };
@ -286,7 +310,7 @@ impl<'a> System<'a> for Sys {
comp_sync_package.comp_removed::<Ori>(uid); comp_sync_package.comp_removed::<Ori>(uid);
} }
send_msg( send_general(
ServerGeneral::CompSync(comp_sync_package), ServerGeneral::CompSync(comp_sync_package),
entity, entity,
pos, pos,
@ -299,19 +323,20 @@ impl<'a> System<'a> for Sys {
// Handle entity deletion in regions that don't exist in RegionMap // Handle entity deletion in regions that don't exist in RegionMap
// (theoretically none) // (theoretically none)
for (region_key, deleted) in deleted_entities.take_remaining_deleted() { for (region_key, deleted) in deleted_entities.take_remaining_deleted() {
for client in for general_stream in (&mut clients, &subscriptions, &mut general_streams)
(&mut clients, &subscriptions) .join()
.join() .filter_map(|(client, subscription, general_stream)| {
.filter_map(|(client, subscription)| { if client.in_game.is_some() && subscription.regions.contains(&region_key) {
if client.in_game.is_some() && subscription.regions.contains(&region_key) { Some(general_stream)
Some(client) } else {
} else { None
None }
} })
})
{ {
for uid in &deleted { for uid in &deleted {
client.send_msg(ServerGeneral::DeleteEntity(Uid(*uid))); let _ = general_stream
.0
.send(ServerGeneral::DeleteEntity(Uid(*uid)));
} }
} }
} }
@ -319,15 +344,19 @@ impl<'a> System<'a> for Sys {
// TODO: Sync clients that don't have a position? // TODO: Sync clients that don't have a position?
// Sync inventories // Sync inventories
for (client, inventory, update) in (&mut clients, &inventories, &inventory_updates).join() { for (inventory, update, in_game_stream) in
client.send_msg(ServerGeneral::InventoryUpdate( (&inventories, &inventory_updates, &mut in_game_streams).join()
{
let _ = in_game_stream.0.send(ServerGeneral::InventoryUpdate(
inventory.clone(), inventory.clone(),
update.event(), update.event(),
)); ));
} }
// Sync outcomes // Sync outcomes
for (client, player, pos) in (&mut clients, &players, positions.maybe()).join() { for (player, pos, in_game_stream) in
(&players, positions.maybe(), &mut in_game_streams).join()
{
let is_near = |o_pos: Vec3<f32>| { let is_near = |o_pos: Vec3<f32>| {
pos.zip_with(player.view_distance, |pos, vd| { pos.zip_with(player.view_distance, |pos, vd| {
pos.0.xy().distance_squared(o_pos.xy()) pos.0.xy().distance_squared(o_pos.xy())
@ -341,7 +370,7 @@ impl<'a> System<'a> for Sys {
.cloned() .cloned()
.collect::<Vec<_>>(); .collect::<Vec<_>>();
if !outcomes.is_empty() { if !outcomes.is_empty() {
client.send_msg(ServerGeneral::Outcomes(outcomes)); let _ = in_game_stream.0.send(ServerGeneral::Outcomes(outcomes));
} }
} }
outcomes.clear(); outcomes.clear();
@ -354,8 +383,8 @@ impl<'a> System<'a> for Sys {
// TODO: doesn't really belong in this system (rename system or create another // TODO: doesn't really belong in this system (rename system or create another
// system?) // system?)
let tof_msg = ServerGeneral::TimeOfDay(*time_of_day); let tof_msg = ServerGeneral::TimeOfDay(*time_of_day);
for client in (&mut clients).join() { for general_stream in (&mut general_streams).join() {
client.send_msg(tof_msg.clone()); let _ = general_stream.0.send(tof_msg.clone());
} }
timer.end(); timer.end();

17
server/src/sys/invite_timeout.rs
View File

@ -1,5 +1,5 @@
use super::SysTimer; use super::SysTimer;
use crate::client::Client; use crate::client::InGameStream;
use common::{ use common::{
comp::group::{Invite, PendingInvites}, comp::group::{Invite, PendingInvites},
msg::{InviteAnswer, ServerGeneral}, msg::{InviteAnswer, ServerGeneral},
@ -16,14 +16,14 @@ impl<'a> System<'a> for Sys {
Entities<'a>, Entities<'a>,
WriteStorage<'a, Invite>, WriteStorage<'a, Invite>,
WriteStorage<'a, PendingInvites>, WriteStorage<'a, PendingInvites>,
WriteStorage<'a, Client>, WriteStorage<'a, InGameStream>,
ReadStorage<'a, Uid>, ReadStorage<'a, Uid>,
Write<'a, SysTimer<Self>>, Write<'a, SysTimer<Self>>,
); );
fn run( fn run(
&mut self, &mut self,
(entities, mut invites, mut pending_invites, mut clients, uids, mut timer): Self::SystemData, (entities, mut invites, mut pending_invites, mut in_game_streams, uids, mut timer): Self::SystemData,
) { ) {
span!(_guard, "run", "invite_timeout::Sys::run"); span!(_guard, "run", "invite_timeout::Sys::run");
timer.start(); timer.start();
@ -51,13 +51,14 @@ impl<'a> System<'a> for Sys {
} }
// Inform inviter of timeout // Inform inviter of timeout
if let (Some(client), Some(target)) = if let (Some(in_game_stream), Some(target)) = (
(clients.get_mut(*inviter), uids.get(invitee).copied()) in_game_streams.get_mut(*inviter),
{ uids.get(invitee).copied(),
client.send_msg(ServerGeneral::InviteComplete { ) {
let _ = in_game_stream.0.send(ServerGeneral::InviteComplete {
target, target,
answer: InviteAnswer::TimedOut, answer: InviteAnswer::TimedOut,
}) });
} }
Some(invitee) Some(invitee)

98
server/src/sys/message.rs
View File

@ -2,7 +2,9 @@ use super::SysTimer;
use crate::{ use crate::{
alias_validator::AliasValidator, alias_validator::AliasValidator,
character_creator, character_creator,
client::Client, client::{
CharacterScreenStream, Client, GeneralStream, InGameStream, PingStream, RegisterStream,
},
login_provider::LoginProvider, login_provider::LoginProvider,
metrics::{NetworkRequestMetrics, PlayerMetrics}, metrics::{NetworkRequestMetrics, PlayerMetrics},
persistence::character_loader::CharacterLoader, persistence::character_loader::CharacterLoader,
@ -41,6 +43,7 @@ impl Sys {
new_chat_msgs: &mut Vec<(Option<specs::Entity>, UnresolvedChatMsg)>, new_chat_msgs: &mut Vec<(Option<specs::Entity>, UnresolvedChatMsg)>,
entity: specs::Entity, entity: specs::Entity,
client: &mut Client, client: &mut Client,
general_stream: &mut GeneralStream,
player_metrics: &ReadExpect<'_, PlayerMetrics>, player_metrics: &ReadExpect<'_, PlayerMetrics>,
uids: &ReadStorage<'_, Uid>, uids: &ReadStorage<'_, Uid>,
chat_modes: &ReadStorage<'_, ChatMode>, chat_modes: &ReadStorage<'_, ChatMode>,
@ -68,7 +71,9 @@ impl Sys {
} }
}, },
ClientGeneral::Disconnect => { ClientGeneral::Disconnect => {
client.send_msg(ServerGeneral::Disconnect(DisconnectReason::Requested)); general_stream
.0
.send(ServerGeneral::Disconnect(DisconnectReason::Requested))?;
}, },
ClientGeneral::Terminate => { ClientGeneral::Terminate => {
debug!(?entity, "Client send message to termitate session"); debug!(?entity, "Client send message to termitate session");
@ -88,6 +93,7 @@ impl Sys {
server_emitter: &mut common::event::Emitter<'_, ServerEvent>, server_emitter: &mut common::event::Emitter<'_, ServerEvent>,
entity: specs::Entity, entity: specs::Entity,
client: &mut Client, client: &mut Client,
in_game_stream: &mut InGameStream,
terrain: &ReadExpect<'_, TerrainGrid>, terrain: &ReadExpect<'_, TerrainGrid>,
network_metrics: &ReadExpect<'_, NetworkRequestMetrics>, network_metrics: &ReadExpect<'_, NetworkRequestMetrics>,
can_build: &ReadStorage<'_, CanBuild>, can_build: &ReadStorage<'_, CanBuild>,
@ -115,7 +121,7 @@ impl Sys {
ClientGeneral::ExitInGame => { ClientGeneral::ExitInGame => {
client.in_game = None; client.in_game = None;
server_emitter.emit(ServerEvent::ExitIngame { entity }); server_emitter.emit(ServerEvent::ExitIngame { entity });
client.send_msg(ServerGeneral::ExitInGameSuccess); in_game_stream.0.send(ServerGeneral::ExitInGameSuccess)?;
}, },
ClientGeneral::SetViewDistance(view_distance) => { ClientGeneral::SetViewDistance(view_distance) => {
players.get_mut(entity).map(|player| { players.get_mut(entity).map(|player| {
@ -133,9 +139,9 @@ impl Sys {
.map(|max| view_distance > max) .map(|max| view_distance > max)
.unwrap_or(false) .unwrap_or(false)
{ {
client.send_msg(ServerGeneral::SetViewDistance( in_game_stream.0.send(ServerGeneral::SetViewDistance(
settings.max_view_distance.unwrap_or(0), settings.max_view_distance.unwrap_or(0),
)); ))?;
} }
}, },
ClientGeneral::ControllerInputs(inputs) => { ClientGeneral::ControllerInputs(inputs) => {
@ -203,10 +209,10 @@ impl Sys {
match terrain.get_key(key) { match terrain.get_key(key) {
Some(chunk) => { Some(chunk) => {
network_metrics.chunks_served_from_memory.inc(); network_metrics.chunks_served_from_memory.inc();
client.send_msg(ServerGeneral::TerrainChunkUpdate { in_game_stream.0.send(ServerGeneral::TerrainChunkUpdate {
key, key,
chunk: Ok(Box::new(chunk.clone())), chunk: Ok(Box::new(chunk.clone())),
}) })?
}, },
None => { None => {
network_metrics.chunks_generation_triggered.inc(); network_metrics.chunks_generation_triggered.inc();
@ -243,6 +249,7 @@ impl Sys {
new_chat_msgs: &mut Vec<(Option<specs::Entity>, UnresolvedChatMsg)>, new_chat_msgs: &mut Vec<(Option<specs::Entity>, UnresolvedChatMsg)>,
entity: specs::Entity, entity: specs::Entity,
client: &mut Client, client: &mut Client,
character_screen_stream: &mut CharacterScreenStream,
character_loader: &ReadExpect<'_, CharacterLoader>, character_loader: &ReadExpect<'_, CharacterLoader>,
uids: &ReadStorage<'_, Uid>, uids: &ReadStorage<'_, Uid>,
players: &mut WriteStorage<'_, Player>, players: &mut WriteStorage<'_, Player>,
@ -278,10 +285,11 @@ impl Sys {
// Give the player a welcome message // Give the player a welcome message
if !editable_settings.server_description.is_empty() { if !editable_settings.server_description.is_empty() {
client.send_msg( character_screen_stream
ChatType::CommandInfo .0
.server_msg(String::from(&*editable_settings.server_description)), .send(ChatType::CommandInfo.server_msg(String::from(
); &*editable_settings.server_description,
)))?;
} }
if !client.login_msg_sent { if !client.login_msg_sent {
@ -295,9 +303,11 @@ impl Sys {
} }
} }
} else { } else {
client.send_msg(ServerGeneral::CharacterDataLoadError(String::from( character_screen_stream
"Failed to fetch player entity", .0
))) .send(ServerGeneral::CharacterDataLoadError(String::from(
"Failed to fetch player entity",
)))?
} }
} }
ClientGeneral::Character(_) => { ClientGeneral::Character(_) => {
@ -313,7 +323,9 @@ impl Sys {
ClientGeneral::CreateCharacter { alias, tool, body } => { ClientGeneral::CreateCharacter { alias, tool, body } => {
if let Err(error) = alias_validator.validate(&alias) { if let Err(error) = alias_validator.validate(&alias) {
debug!(?error, ?alias, "denied alias as it contained a banned word"); debug!(?error, ?alias, "denied alias as it contained a banned word");
client.send_msg(ServerGeneral::CharacterActionError(error.to_string())); character_screen_stream
.0
.send(ServerGeneral::CharacterActionError(error.to_string()))?;
} else if let Some(player) = players.get(entity) { } else if let Some(player) = players.get(entity) {
character_creator::create_character( character_creator::create_character(
entity, entity,
@ -340,9 +352,13 @@ impl Sys {
} }
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
fn handle_ping_msg(client: &mut Client, msg: PingMsg) -> Result<(), crate::error::Error> { fn handle_ping_msg(
client: &mut Client,
ping_stream: &mut PingStream,
msg: PingMsg,
) -> Result<(), crate::error::Error> {
match msg { match msg {
PingMsg::Ping => client.send_msg(PingMsg::Pong), PingMsg::Ping => ping_stream.0.send(PingMsg::Pong)?,
PingMsg::Pong => {}, PingMsg::Pong => {},
} }
Ok(()) Ok(())
@ -354,6 +370,7 @@ impl Sys {
new_players: &mut Vec<specs::Entity>, new_players: &mut Vec<specs::Entity>,
entity: specs::Entity, entity: specs::Entity,
client: &mut Client, client: &mut Client,
register_stream: &mut RegisterStream,
player_metrics: &ReadExpect<'_, PlayerMetrics>, player_metrics: &ReadExpect<'_, PlayerMetrics>,
login_provider: &mut WriteExpect<'_, LoginProvider>, login_provider: &mut WriteExpect<'_, LoginProvider>,
admins: &mut WriteStorage<'_, Admin>, admins: &mut WriteStorage<'_, Admin>,
@ -368,9 +385,7 @@ impl Sys {
&*editable_settings.banlist, &*editable_settings.banlist,
) { ) {
Err(err) => { Err(err) => {
client register_stream.0.send(ServerRegisterAnswer::Err(err))?;
.register_stream
.send(ServerRegisterAnswer::Err(err))?;
return Ok(()); return Ok(());
}, },
Ok((username, uuid)) => (username, uuid), Ok((username, uuid)) => (username, uuid),
@ -382,8 +397,8 @@ impl Sys {
if !player.is_valid() { if !player.is_valid() {
// Invalid player // Invalid player
client register_stream
.register_stream .0
.send(ServerRegisterAnswer::Err(RegisterError::InvalidCharacter))?; .send(ServerRegisterAnswer::Err(RegisterError::InvalidCharacter))?;
return Ok(()); return Ok(());
} }
@ -401,12 +416,14 @@ impl Sys {
// Tell the client its request was successful. // Tell the client its request was successful.
client.registered = true; client.registered = true;
client.register_stream.send(ServerRegisterAnswer::Ok(()))?; register_stream.0.send(ServerRegisterAnswer::Ok(()))?;
// Send initial player list // Send initial player list
client.send_msg(ServerGeneral::PlayerListUpdate(PlayerListUpdate::Init( register_stream
player_list.clone(), .0
))); .send(ServerGeneral::PlayerListUpdate(PlayerListUpdate::Init(
player_list.clone(),
)))?;
// Add to list to notify all clients of the new player // Add to list to notify all clients of the new player
new_players.push(entity); new_players.push(entity);
@ -446,19 +463,8 @@ impl Sys {
editable_settings: &ReadExpect<'_, EditableSettings>, editable_settings: &ReadExpect<'_, EditableSettings>,
alias_validator: &ReadExpect<'_, AliasValidator>, alias_validator: &ReadExpect<'_, AliasValidator>,
) -> Result<(), crate::error::Error> { ) -> Result<(), crate::error::Error> {
let (mut b1, mut b2, mut b3, mut b4, mut b5) = (
client.network_error,
client.network_error,
client.network_error,
client.network_error,
client.network_error,
);
loop { loop {
/* /*
waiting for 1 of the 5 streams to return a massage asynchronous.
If so, handle that msg type. This code will be refactored soon
*/
let q1 = Client::internal_recv(&mut b1, &mut client.general_stream); let q1 = Client::internal_recv(&mut b1, &mut client.general_stream);
let q2 = Client::internal_recv(&mut b2, &mut client.in_game_stream); let q2 = Client::internal_recv(&mut b2, &mut client.in_game_stream);
let q3 = Client::internal_recv(&mut b3, &mut client.character_screen_stream); let q3 = Client::internal_recv(&mut b3, &mut client.character_screen_stream);
@ -540,7 +546,7 @@ impl Sys {
editable_settings, editable_settings,
msg?, msg?,
)?; )?;
} }*/
} }
} }
} }
@ -571,6 +577,11 @@ impl<'a> System<'a> for Sys {
WriteStorage<'a, Ori>, WriteStorage<'a, Ori>,
WriteStorage<'a, Player>, WriteStorage<'a, Player>,
WriteStorage<'a, Client>, WriteStorage<'a, Client>,
WriteStorage<'a, GeneralStream>,
//WriteStorage<'a, PingStream>,
//WriteStorage<'a, RegisterStream>,
//WriteStorage<'a, CharacterScreenStream>,
//WriteStorage<'a, InGameStream>,
WriteStorage<'a, Controller>, WriteStorage<'a, Controller>,
Read<'a, Settings>, Read<'a, Settings>,
ReadExpect<'a, EditableSettings>, ReadExpect<'a, EditableSettings>,
@ -604,6 +615,11 @@ impl<'a> System<'a> for Sys {
mut orientations, mut orientations,
mut players, mut players,
mut clients, mut clients,
mut general_streams,
//mut ping_streams,
//mut register_streams,
//mut character_screen_streams,
//mut in_game_streams,
mut controllers, mut controllers,
settings, settings,
editable_settings, editable_settings,
@ -697,7 +713,8 @@ impl<'a> System<'a> for Sys {
server_emitter.emit(ServerEvent::ClientDisconnect(entity)); server_emitter.emit(ServerEvent::ClientDisconnect(entity));
} else if time.0 - client.last_ping > settings.client_timeout.as_secs() as f64 * 0.5 { } else if time.0 - client.last_ping > settings.client_timeout.as_secs() as f64 * 0.5 {
// Try pinging the client if the timeout is nearing. // Try pinging the client if the timeout is nearing.
client.send_msg(PingMsg::Ping); //FIXME
//client.send_msg(PingMsg::Ping);
} }
} }
@ -713,7 +730,8 @@ impl<'a> System<'a> for Sys {
character: None, // new players will be on character select. character: None, // new players will be on character select.
})); }));
for client in (&mut clients).join().filter(|c| c.registered) { for client in (&mut clients).join().filter(|c| c.registered) {
client.send_msg(msg.clone()) //FIXME
//client.send_msg(msg.clone())
} }
} }
} }

35
server/src/sys/subscription.rs
View File

@ -2,7 +2,7 @@ use super::{
sentinel::{DeletedEntities, TrackedComps}, sentinel::{DeletedEntities, TrackedComps},
SysTimer, SysTimer,
}; };
use crate::client::{self, Client, RegionSubscription}; use crate::client::{self, Client, InGameStream, RegionSubscription};
use common::{ use common::{
comp::{Ori, Player, Pos, Vel}, comp::{Ori, Player, Pos, Vel},
msg::ServerGeneral, msg::ServerGeneral,
@ -32,7 +32,8 @@ impl<'a> System<'a> for Sys {
ReadStorage<'a, Vel>, ReadStorage<'a, Vel>,
ReadStorage<'a, Ori>, ReadStorage<'a, Ori>,
ReadStorage<'a, Player>, ReadStorage<'a, Player>,
WriteStorage<'a, Client>, ReadStorage<'a, Client>,
WriteStorage<'a, InGameStream>,
WriteStorage<'a, RegionSubscription>, WriteStorage<'a, RegionSubscription>,
Write<'a, DeletedEntities>, Write<'a, DeletedEntities>,
TrackedComps<'a>, TrackedComps<'a>,
@ -50,7 +51,8 @@ impl<'a> System<'a> for Sys {
velocities, velocities,
orientations, orientations,
players, players,
mut clients, clients,
mut in_game_streams,
mut subscriptions, mut subscriptions,
mut deleted_entities, mut deleted_entities,
tracked_comps, tracked_comps,
@ -71,17 +73,18 @@ impl<'a> System<'a> for Sys {
// 7. Determine list of regions that are in range and iterate through it // 7. Determine list of regions that are in range and iterate through it
// - check if in hashset (hash calc) if not add it // - check if in hashset (hash calc) if not add it
let mut regions_to_remove = Vec::new(); let mut regions_to_remove = Vec::new();
for (client, subscription, pos, vd, client_entity) in ( for (_, subscription, pos, vd, client_entity, in_game_stream) in (
&mut clients, &clients,
&mut subscriptions, &mut subscriptions,
&positions, &positions,
&players, &players,
&entities, &entities,
&mut in_game_streams,
) )
.join() .join()
.filter_map(|(client, s, pos, player, e)| { .filter_map(|(client, s, pos, player, e, stream)| {
if client.in_game.is_some() { if client.in_game.is_some() {
player.view_distance.map(|v| (client, s, pos, v, e)) player.view_distance.map(|v| (client, s, pos, v, e, stream))
} else { } else {
None None
} }
@ -153,7 +156,9 @@ impl<'a> System<'a> for Sys {
.map(|key| subscription.regions.contains(key)) .map(|key| subscription.regions.contains(key))
.unwrap_or(false) .unwrap_or(false)
{ {
client.send_msg(ServerGeneral::DeleteEntity(uid)); let _ = in_game_stream
.0
.send(ServerGeneral::DeleteEntity(uid));
} }
} }
}, },
@ -161,7 +166,7 @@ impl<'a> System<'a> for Sys {
} }
// Tell client to delete entities in the region // Tell client to delete entities in the region
for (&uid, _) in (&uids, region.entities()).join() { for (&uid, _) in (&uids, region.entities()).join() {
client.send_msg(ServerGeneral::DeleteEntity(uid)); let _ = in_game_stream.0.send(ServerGeneral::DeleteEntity(uid));
} }
} }
// Send deleted entities since they won't be processed for this client in entity // Send deleted entities since they won't be processed for this client in entity
@ -171,7 +176,9 @@ impl<'a> System<'a> for Sys {
.iter() .iter()
.flat_map(|v| v.iter()) .flat_map(|v| v.iter())
{ {
client.send_msg(ServerGeneral::DeleteEntity(Uid(*uid))); let _ = in_game_stream
.0
.send(ServerGeneral::DeleteEntity(Uid(*uid)));
} }
} }
@ -196,7 +203,7 @@ impl<'a> System<'a> for Sys {
{ {
// Send message to create entity and tracked components and physics // Send message to create entity and tracked components and physics
// components // components
client.send_msg(ServerGeneral::CreateEntity( let _ = in_game_stream.0.send(ServerGeneral::CreateEntity(
tracked_comps.create_entity_package( tracked_comps.create_entity_package(
entity, entity,
Some(*pos), Some(*pos),
@ -217,14 +224,14 @@ impl<'a> System<'a> for Sys {
/// Initialize region subscription /// Initialize region subscription
pub fn initialize_region_subscription(world: &World, entity: specs::Entity) { pub fn initialize_region_subscription(world: &World, entity: specs::Entity) {
if let (Some(client_pos), Some(client_vd), Some(client)) = ( if let (Some(client_pos), Some(client_vd), Some(in_game_stream)) = (
world.read_storage::<Pos>().get(entity), world.read_storage::<Pos>().get(entity),
world world
.read_storage::<Player>() .read_storage::<Player>()
.get(entity) .get(entity)
.map(|pl| pl.view_distance) .map(|pl| pl.view_distance)
.and_then(|v| v), .and_then(|v| v),
world.write_storage::<Client>().get_mut(entity), world.write_storage::<InGameStream>().get_mut(entity),
) { ) {
let fuzzy_chunk = (Vec2::<f32>::from(client_pos.0)) let fuzzy_chunk = (Vec2::<f32>::from(client_pos.0))
.map2(TerrainChunkSize::RECT_SIZE, |e, sz| e as i32 / sz as i32); .map2(TerrainChunkSize::RECT_SIZE, |e, sz| e as i32 / sz as i32);
@ -249,7 +256,7 @@ pub fn initialize_region_subscription(world: &World, entity: specs::Entity) {
.join() .join()
{ {
// Send message to create entity and tracked components and physics components // Send message to create entity and tracked components and physics components
client.send_msg(ServerGeneral::CreateEntity( let _ = in_game_stream.0.send(ServerGeneral::CreateEntity(
tracked_comps.create_entity_package( tracked_comps.create_entity_package(
entity, entity,
Some(*pos), Some(*pos),

18
server/src/sys/terrain.rs
View File

@ -1,5 +1,5 @@
use super::SysTimer; use super::SysTimer;
use crate::{chunk_generator::ChunkGenerator, client::Client, Tick}; use crate::{chunk_generator::ChunkGenerator, client::InGameStream, Tick};
use common::{ use common::{
comp::{self, bird_medium, Alignment, Player, Pos}, comp::{self, bird_medium, Alignment, Player, Pos},
event::{EventBus, ServerEvent}, event::{EventBus, ServerEvent},
@ -34,7 +34,7 @@ impl<'a> System<'a> for Sys {
Write<'a, TerrainChanges>, Write<'a, TerrainChanges>,
ReadStorage<'a, Pos>, ReadStorage<'a, Pos>,
ReadStorage<'a, Player>, ReadStorage<'a, Player>,
WriteStorage<'a, Client>, WriteStorage<'a, InGameStream>,
); );
fn run( fn run(
@ -48,7 +48,7 @@ impl<'a> System<'a> for Sys {
mut terrain_changes, mut terrain_changes,
positions, positions,
players, players,
mut clients, mut in_game_streams,
): Self::SystemData, ): Self::SystemData,
) { ) {
span!(_guard, "run", "terrain::Sys::run"); span!(_guard, "run", "terrain::Sys::run");
@ -62,8 +62,8 @@ impl<'a> System<'a> for Sys {
let (chunk, supplement) = match res { let (chunk, supplement) = match res {
Ok((chunk, supplement)) => (chunk, supplement), Ok((chunk, supplement)) => (chunk, supplement),
Err(Some(entity)) => { Err(Some(entity)) => {
if let Some(client) = clients.get_mut(entity) { if let Some(in_game_stream) = in_game_streams.get_mut(entity) {
client.send_msg(ServerGeneral::TerrainChunkUpdate { let _ = in_game_stream.0.send(ServerGeneral::TerrainChunkUpdate {
key, key,
chunk: Err(()), chunk: Err(()),
}); });
@ -75,10 +75,10 @@ impl<'a> System<'a> for Sys {
}, },
}; };
// Send the chunk to all nearby players. // Send the chunk to all nearby players.
for (view_distance, pos, client) in (&players, &positions, &mut clients) for (view_distance, pos, in_game_stream) in (&players, &positions, &mut in_game_streams)
.join() .join()
.filter_map(|(player, pos, client)| { .filter_map(|(player, pos, in_game_stream)| {
player.view_distance.map(|vd| (vd, pos, client)) player.view_distance.map(|vd| (vd, pos, in_game_stream))
}) })
{ {
let chunk_pos = terrain.pos_key(pos.0.map(|e| e as i32)); let chunk_pos = terrain.pos_key(pos.0.map(|e| e as i32));
@ -90,7 +90,7 @@ impl<'a> System<'a> for Sys {
.magnitude_squared(); .magnitude_squared();
if adjusted_dist_sqr <= view_distance.pow(2) { if adjusted_dist_sqr <= view_distance.pow(2) {
client.send_msg(ServerGeneral::TerrainChunkUpdate { let _ = in_game_stream.0.send(ServerGeneral::TerrainChunkUpdate {
key, key,
chunk: Ok(Box::new(chunk.clone())), chunk: Ok(Box::new(chunk.clone())),
}); });

15
server/src/sys/terrain_sync.rs
View File

@ -1,5 +1,5 @@
use super::SysTimer; use super::SysTimer;
use crate::client::Client; use crate::client::InGameStream;
use common::{ use common::{
comp::{Player, Pos}, comp::{Player, Pos},
msg::ServerGeneral, msg::ServerGeneral,
@ -20,25 +20,26 @@ impl<'a> System<'a> for Sys {
Write<'a, SysTimer<Self>>, Write<'a, SysTimer<Self>>,
ReadStorage<'a, Pos>, ReadStorage<'a, Pos>,
ReadStorage<'a, Player>, ReadStorage<'a, Player>,
WriteStorage<'a, Client>, WriteStorage<'a, InGameStream>,
); );
fn run( fn run(
&mut self, &mut self,
(terrain, terrain_changes, mut timer, positions, players, mut clients): Self::SystemData, (terrain, terrain_changes, mut timer, positions, players, mut in_game_streams): Self::SystemData,
) { ) {
span!(_guard, "run", "terrain_sync::Sys::run"); span!(_guard, "run", "terrain_sync::Sys::run");
timer.start(); timer.start();
// Sync changed chunks // Sync changed chunks
'chunk: for chunk_key in &terrain_changes.modified_chunks { 'chunk: for chunk_key in &terrain_changes.modified_chunks {
for (player, pos, client) in (&players, &positions, &mut clients).join() { for (player, pos, in_game_stream) in (&players, &positions, &mut in_game_streams).join()
{
if player if player
.view_distance .view_distance
.map(|vd| super::terrain::chunk_in_vd(pos.0, *chunk_key, &terrain, vd)) .map(|vd| super::terrain::chunk_in_vd(pos.0, *chunk_key, &terrain, vd))
.unwrap_or(false) .unwrap_or(false)
{ {
client.send_msg(ServerGeneral::TerrainChunkUpdate { let _ = in_game_stream.0.send(ServerGeneral::TerrainChunkUpdate {
key: *chunk_key, key: *chunk_key,
chunk: Ok(Box::new(match terrain.get_key(*chunk_key) { chunk: Ok(Box::new(match terrain.get_key(*chunk_key) {
Some(chunk) => chunk.clone(), Some(chunk) => chunk.clone(),
@ -52,9 +53,9 @@ impl<'a> System<'a> for Sys {
// TODO: Don't send all changed blocks to all clients // TODO: Don't send all changed blocks to all clients
// Sync changed blocks // Sync changed blocks
let msg = ServerGeneral::TerrainBlockUpdates(terrain_changes.modified_blocks.clone()); let msg = ServerGeneral::TerrainBlockUpdates(terrain_changes.modified_blocks.clone());
for (player, client) in (&players, &mut clients).join() { for (player, in_game_stream) in (&players, &mut in_game_streams).join() {
if player.view_distance.is_some() { if player.view_distance.is_some() {
client.send_msg(msg.clone()); let _ = in_game_stream.0.send(msg.clone());
} }
} }

24
server/src/sys/waypoint.rs
View File

@ -1,5 +1,5 @@
use super::SysTimer; use super::SysTimer;
use crate::client::Client; use crate::client::GeneralStream;
use common::{ use common::{
comp::{Player, Pos, Waypoint, WaypointArea}, comp::{Player, Pos, Waypoint, WaypointArea},
msg::{Notification, ServerGeneral}, msg::{Notification, ServerGeneral},
@ -22,28 +22,38 @@ impl<'a> System<'a> for Sys {
ReadStorage<'a, Player>, ReadStorage<'a, Player>,
ReadStorage<'a, WaypointArea>, ReadStorage<'a, WaypointArea>,
WriteStorage<'a, Waypoint>, WriteStorage<'a, Waypoint>,
WriteStorage<'a, Client>, WriteStorage<'a, GeneralStream>,
Read<'a, Time>, Read<'a, Time>,
Write<'a, SysTimer<Self>>, Write<'a, SysTimer<Self>>,
); );
fn run( fn run(
&mut self, &mut self,
(entities, positions, players, waypoint_areas, mut waypoints, mut clients, time, mut timer): Self::SystemData, (
entities,
positions,
players,
waypoint_areas,
mut waypoints,
mut general_streams,
time,
mut timer,
): Self::SystemData,
) { ) {
span!(_guard, "run", "waypoint::Sys::run"); span!(_guard, "run", "waypoint::Sys::run");
timer.start(); timer.start();
for (entity, player_pos, _, client) in for (entity, player_pos, _, general_stream) in
(&entities, &positions, &players, &mut clients).join() (&entities, &positions, &players, &mut general_streams).join()
{ {
for (waypoint_pos, waypoint_area) in (&positions, &waypoint_areas).join() { for (waypoint_pos, waypoint_area) in (&positions, &waypoint_areas).join() {
if player_pos.0.distance_squared(waypoint_pos.0) < waypoint_area.radius().powi(2) { if player_pos.0.distance_squared(waypoint_pos.0) < waypoint_area.radius().powi(2) {
if let Ok(wp_old) = waypoints.insert(entity, Waypoint::new(player_pos.0, *time)) if let Ok(wp_old) = waypoints.insert(entity, Waypoint::new(player_pos.0, *time))
{ {
if wp_old.map_or(true, |w| w.elapsed(*time) > NOTIFY_TIME) { if wp_old.map_or(true, |w| w.elapsed(*time) > NOTIFY_TIME) {
client let _ = general_stream
.send_msg(ServerGeneral::Notification(Notification::WaypointSaved)); .0
.send(ServerGeneral::Notification(Notification::WaypointSaved));
} }
} }
} }